Small Business: for Customer Support A Deep Dive

// ingestion.mjs - Ticket ingestion endpoint for email webhooks and public forms
// Dependencies: fastify@4.28.0, @fastify/rate-limit@8.0.0, ioredis@5.3.0, zod@3.22.0, ulid@2.3.0
import Fastify from 'fastify';
import rateLimit from '@fastify/rate-limit';
import Redis from 'ioredis';
import { z } from 'zod';
import { ulid } from 'ulid';
import dotenv from 'dotenv';

dotenv.config();

// Initialize Fastify with logger for production debugging
const fastify = Fastify({
  logger: process.env.NODE_ENV === 'production',
  ignoreTrailingSlash: true
});

// Configure rate limiting to prevent abuse: 100 requests per 15 minutes per IP
await fastify.register(rateLimit, {
  max: 100,
  timeWindow: '15 minutes',
  errorResponseBuilder: (req, context) => ({
    statusCode: 429,
    error: 'Too Many Requests',
    message: `Rate limit exceeded. Try again in ${context.ttl} seconds.`
  })
});

// Initialize Redis client for publishing to ticket ingestion stream
const redis = new Redis({
  host: process.env.REDIS_HOST || 'localhost',
  port: process.env.REDIS_PORT || 6379,
  password: process.env.REDIS_PASSWORD,
  retryStrategy: (times) => Math.min(times * 50, 2000) // Exponential backoff for Redis reconnects
});

// Zod schema for validating incoming ticket payloads (email webhooks + public forms)
const TicketSchema = z.object({
  source: z.enum(['email', 'chat', 'public_form']),
  requesterEmail: z.string().email('Invalid requester email'),
  subject: z.string().min(5, 'Subject must be at least 5 characters').max(200),
  body: z.string().min(10, 'Body must be at least 10 characters').max(5000),
  attachments: z.array(z.string().url()).optional().default([]),
  metadata: z.record(z.unknown()).optional().default({})
});

// Health check endpoint for load balancers
fastify.get('/health', async (req, reply) => {
  try {
    // Verify Redis connectivity
    await redis.ping();
    return { status: 'healthy', redis: 'connected', timestamp: Date.now() };
  } catch (err) {
    reply.code(503).send({ status: 'unhealthy', error: err.message });
  }
});

// Main ingestion endpoint for all ticket sources
fastify.post('/tickets/ingest', async (req, reply) => {
  try {
    // Validate payload against schema
    const validated = TicketSchema.parse(req.body);

    // Generate unique ticket ID (ULID for time-sorted IDs)
    const ticketId = ulid();
    const streamKey = 'ticket:ingestion:stream';

    // Publish to Redis Streams with ticket ID as entry ID
    const streamEntry = [
      'ticketId', ticketId,
      'source', validated.source,
      'requesterEmail', validated.requesterEmail,
      'subject', validated.subject,
      'body', validated.body,
      'attachments', JSON.stringify(validated.attachments),
      'metadata', JSON.stringify(validated.metadata),
      'receivedAt', Date.now().toString()
    ];

    const result = await redis.xadd(streamKey, '*', ...streamEntry);
    if (!result) {
      throw new Error('Failed to publish ticket to Redis Streams');
    }

    // Return 202 Accepted (async processing)
    reply.code(202).send({
      ticketId,
      status: 'queued',
      estimatedProcessingTimeMs: 1500
    });
  } catch (err) {
    // Handle Zod validation errors
    if (err instanceof z.ZodError) {
      return reply.code(400).send({
        error: 'Validation Failed',
        details: err.errors
      });
    }
    // Handle Redis errors
    if (err.message.includes('Redis')) {
      fastify.log.error('Redis publish error:', err);
      return reply.code(503).send({ error: 'Message broker unavailable' });
    }
    // Generic error handling
    fastify.log.error('Ingestion error:', err);
    reply.code(500).send({ error: 'Internal Server Error' });
  }
});

// Start server
const start = async () => {
  try {
    await fastify.listen({
      port: process.env.PORT || 3000,
      host: '0.0.0.0'
    });
    fastify.log.info(`Ingestion service listening on port ${process.env.PORT || 3000}`);
  } catch (err) {
    fastify.log.error(err);
    process.exit(1);
  }
};

start();

// triage-worker.mjs - Consumes ticket ingestion stream, applies triage rules, assigns agents
// Dependencies: ioredis@5.3.0, pg@8.11.0, @xenova/transformers@2.6.0 (local LLM for intent detection)
import Redis from 'ioredis';
import { Pool } from 'pg';
import { pipeline } from '@xenova/transformers';
import dotenv from 'dotenv';

dotenv.config();

// Initialize Redis client for consuming from ingestion stream
const redis = new Redis({
  host: process.env.REDIS_HOST || 'localhost',
  port: process.env.REDIS_PORT || 6379,
  password: process.env.REDIS_PASSWORD
});

// Initialize Postgres pool for ticket storage and agent availability
const pgPool = new Pool({
  host: process.env.POSTGRES_HOST || 'localhost',
  port: process.env.POSTGRES_PORT || 5432,
  database: process.env.POSTGRES_DB || 'support_db',
  user: process.env.POSTGRES_USER || 'support_user',
  password: process.env.POSTGRES_PASSWORD,
  max: 20, // Max connections for triage worker
  idleTimeoutMillis: 30000
});

// Load local LLM for intent detection (distilbert-base-uncased-finetuned-sst-2-english)
// Returns 0 (negative) to 1 (positive) sentiment, used for priority scoring
let intentClassifier;
try {
  intentClassifier = await pipeline('sentiment-analysis', 'Xenova/distilbert-base-uncased-finetuned-sst-2-english');
} catch (err) {
  console.error('Failed to load LLM model:', err);
  process.exit(1);
}

// Triage rules: ordered by priority, first match wins
const TRIAGE_RULES = [
  {
    name: 'billing_urgent',
    condition: (ticket) => ticket.subject.toLowerCase().includes('refund') && ticket.metadata.orderValue > 500,
    priority: 'urgent',
    assignToTeam: 'billing'
  },
  {
    name: 'technical_bug',
    condition: (ticket) => ticket.source === 'chat' && ticket.subject.toLowerCase().includes('bug'),
    priority: 'high',
    assignToTeam: 'technical'
  },
  {
    name: 'general_inquiry',
    condition: () => true, // Default rule
    priority: 'normal',
    assignToTeam: 'general'
  }
];

// Helper to get least loaded agent from a team
async function getAvailableAgent(team) {
  const client = await pgPool.connect();
  try {
    const result = await client.query(`
      SELECT user_id, team, active_tickets
      FROM agents
      WHERE team = $1 AND is_available = true
      ORDER BY active_tickets ASC
      LIMIT 1;
    `, [team]);
    return result.rows[0] || null;
  } catch (err) {
    console.error('Agent lookup error:', err);
    throw err;
  } finally {
    client.release();
  }
}

// Main stream consumer logic
const STREAM_KEY = 'ticket:ingestion:stream';
const CONSUMER_GROUP = 'triage-workers';
const CONSUMER_NAME = `worker-${process.pid}`;

async function initializeConsumerGroup() {
  try {
    // Create consumer group if it doesn't exist, start from beginning of stream
    await redis.xgroup(STREAM_KEY, CONSUMER_GROUP, '$', 'MKSTREAM');
  } catch (err) {
    // Group already exists, ignore error
    if (!err.message.includes('BUSYGROUP')) {
      throw err;
    }
  }
}

async function processTicket(entryId, ticketData) {
  const ticket = {
    ticketId: ticketData.ticketId,
    source: ticketData.source,
    requesterEmail: ticketData.requesterEmail,
    subject: ticketData.subject,
    body: ticketData.body,
    attachments: JSON.parse(ticketData.attachments || '[]'),
    metadata: JSON.parse(ticketData.metadata || '{}'),
    receivedAt: parseInt(ticketData.receivedAt)
  };

  // Apply rule-based triage
  let matchedRule = TRIAGE_RULES.find(rule => rule.condition(ticket));
  matchedRule = matchedRule || TRIAGE_RULES[TRIAGE_RULES.length - 1]; // Fallback to default

  // Enhance priority with LLM sentiment analysis
  let llmPriorityBoost = 0;
  try {
    const sentiment = await intentClassifier(ticket.subject + ' ' + ticket.body.slice(0, 200));
    if (sentiment[0].label === 'POSITIVE' && sentiment[0].score > 0.9) {
      // Positive sentiment with high score = likely happy customer, lower priority
      llmPriorityBoost = -1;
    } else if (sentiment[0].label === 'NEGATIVE' && sentiment[0].score > 0.8) {
      // Negative sentiment = angry customer, boost priority
      llmPriorityBoost = 1;
    }
  } catch (err) {
    console.error('LLM inference error:', err);
    // Fall back to rule-based priority if LLM fails
  }

  // Calculate final priority (urgent: 3, high: 2, normal:1, low:0)
  const priorityMap = { urgent: 3, high: 2, normal: 1, low: 0 };
  let finalPriority = priorityMap[matchedRule.priority] + llmPriorityBoost;
  finalPriority = Math.max(0, Math.min(3, finalPriority)); // Clamp to valid range

  // Assign agent
  const agent = await getAvailableAgent(matchedRule.assignToTeam);
  if (!agent) {
    console.warn(`No available agents for team ${matchedRule.assignToTeam}, queuing ticket`);
  }

  // Persist ticket to Postgres
  const pgClient = await pgPool.connect();
  try {
    await pgClient.query('BEGIN');
    const insertResult = await pgClient.query(`
      INSERT INTO tickets (
        ticket_id, source, requester_email, subject, body, attachments,
        metadata, priority, assigned_team, assigned_agent_id, status, created_at
      ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
      RETURNING id;
    `, [
      ticket.ticketId,
      ticket.source,
      ticket.requesterEmail,
      ticket.subject,
      ticket.body,
      JSON.stringify(ticket.attachments),
      JSON.stringify(ticket.metadata),
      finalPriority,
      matchedRule.assignToTeam,
      agent?.user_id || null,
      'open',
      new Date(ticket.receivedAt)
    ]);

    // Update agent's active ticket count if assigned
    if (agent) {
      await pgClient.query(`
        UPDATE agents SET active_tickets = active_tickets + 1 WHERE user_id = $1;
      `, [agent.user_id]);
    }

    await pgClient.query('COMMIT');
    console.log(`Processed ticket ${ticket.ticketId}, priority ${finalPriority}, assigned to ${agent?.user_id || 'unassigned'}`);
  } catch (err) {
    await pgClient.query('ROLLBACK');
    throw err;
  } finally {
    pgClient.release();
  }

  // Acknowledge message in Redis Streams
  await redis.xack(STREAM_KEY, CONSUMER_GROUP, entryId);
}

// Start consuming from stream
async function startConsumer() {
  await initializeConsumerGroup();
  console.log(`Triage worker ${CONSUMER_NAME} started, listening to ${STREAM_KEY}`);

  while (true) {
    try {
      // Read new messages from stream, block for 5 seconds if no messages
      const streams = await redis.xreadgroup(
        'GROUP', CONSUMER_GROUP, CONSUMER_NAME,
        'COUNT', 10, // Process 10 messages at a time
        'BLOCK', 5000,
        'STREAMS', STREAM_KEY, '>'
      );

      if (!streams || streams.length === 0) continue;

      const [streamKey, messages] = streams[0];
      for (const [entryId, fields] of messages) {
        // Convert flat array of fields to object
        const ticketData = {};
        for (let i = 0; i < fields.length; i += 2) {
          ticketData[fields[i]] = fields[i + 1];
        }
        try {
          await processTicket(entryId, ticketData);
        } catch (err) {
          console.error(`Failed to process entry ${entryId}:`, err);
          // Move failed message to dead letter queue
          await redis.xadd('ticket:dead_letter_queue', '*', ...Object.entries(ticketData).flat(), 'error', err.message);
          await redis.xack(STREAM_KEY, CONSUMER_GROUP, entryId);
        }
      }
    } catch (err) {
      console.error('Stream read error:', err);
      await new Promise(resolve => setTimeout(resolve, 1000)); // Wait 1s before retrying
    }
  }
}

startConsumer();

// dashboard-api.mjs - Agent dashboard API: ticket listing, updates, agent status
// Dependencies: fastify@4.28.0, ioredis@5.3.0, pg@8.11.0, zod@3.22.0
import Fastify from 'fastify';
import Redis from 'ioredis';
import { Pool } from 'pg';
import { z } from 'zod';
import dotenv from 'dotenv';

dotenv.config();

const fastify = Fastify({ logger: true });

// Initialize Redis for caching ticket lists (TTL 30s to balance freshness and load)
const redis = new Redis({
  host: process.env.REDIS_HOST || 'localhost',
  port: process.env.REDIS_PORT || 6379,
  password: process.env.REDIS_PASSWORD
});

// Initialize Postgres pool
const pgPool = new Pool({
  host: process.env.POSTGRES_HOST || 'localhost',
  port: process.env.POSTGRES_PORT || 5432,
  database: process.env.POSTGRES_DB || 'support_db',
  user: process.env.POSTGRES_USER || 'support_user',
  password: process.env.POSTGRES_PASSWORD,
  max: 50
});

// Zod schema for ticket update payload
const TicketUpdateSchema = z.object({
  status: z.enum(['open', 'in_progress', 'resolved', 'closed']).optional(),
  priority: z.number().min(0).max(3).optional(),
  assignedAgentId: z.string().uuid().optional(),
  notes: z.string().max(2000).optional()
});

// Middleware to validate agent JWT (simplified for example)
fastify.addHook('onRequest', async (req, reply) => {
  const authHeader = req.headers.authorization;
  if (!authHeader || !authHeader.startsWith('Bearer ')) {
    reply.code(401).send({ error: 'Unauthorized' });
    return;
  }
  const token = authHeader.split(' ')[1];
  try {
    // In production, verify JWT with your auth provider (e.g., Auth0, Keycloak)
    const agent = await redis.get(`agent:token:${token}`);
    if (!agent) {
      reply.code(401).send({ error: 'Invalid or expired token' });
      return;
    }
    req.agent = JSON.parse(agent);
  } catch (err) {
    reply.code(401).send({ error: 'Unauthorized' });
  }
});

// Get tickets for authenticated agent (with caching)
fastify.get('/api/tickets', async (req, reply) => {
  const agentId = req.agent.user_id;
  const cacheKey = `agent:${agentId}:tickets:${req.query.status || 'all'}`;

  // Check cache first
  try {
    const cached = await redis.get(cacheKey);
    if (cached) {
      return JSON.parse(cached);
    }
  } catch (err) {
    fastify.log.warn('Redis cache read error:', err);
  }

  // Build query based on filters
  const { status, priority, team } = req.query;
  let query = `
    SELECT t.ticket_id, t.source, t.requester_email, t.subject, t.priority,
           t.status, t.assigned_team, t.created_at, a.name as agent_name
    FROM tickets t
    LEFT JOIN agents a ON t.assigned_agent_id = a.user_id
    WHERE 1=1
  `;
  const params = [];
  let paramCount = 1;

  if (status) {
    query += ` AND t.status = $${paramCount}`;
    params.push(status);
    paramCount++;
  }
  if (priority) {
    query += ` AND t.priority = $${paramCount}`;
    params.push(parseInt(priority));
    paramCount++;
  }
  if (team) {
    query += ` AND t.assigned_team = $${paramCount}`;
    params.push(team);
    paramCount++;
  }
  // Agents only see tickets assigned to them or their team if they're a lead
  if (!req.agent.is_team_lead) {
    query += ` AND t.assigned_agent_id = $${paramCount}`;
    params.push(agentId);
    paramCount++;
  } else {
    query += ` AND t.assigned_team = $${paramCount}`;
    params.push(req.agent.team);
    paramCount++;
  }

  query += ` ORDER BY t.priority DESC, t.created_at ASC LIMIT 50;`;

  try {
    const client = await pgPool.connect();
    const result = await client.query(query, params);
    client.release();

    // Cache result for 30 seconds
    await redis.setex(cacheKey, 30, JSON.stringify(result.rows));

    return result.rows;
  } catch (err) {
    fastify.log.error('Ticket list query error:', err);
    reply.code(500).send({ error: 'Failed to fetch tickets' });
  }
});

// Update a ticket (with optimistic locking to prevent overwrites)
fastify.patch('/api/tickets/:ticketId', async (req, reply) => {
  const { ticketId } = req.params;
  try {
    const updates = TicketUpdateSchema.parse(req.body);
    const client = await pgPool.connect();
    try {
      await client.query('BEGIN');

      // Get current ticket version for optimistic locking
      const current = await client.query(`
        SELECT version FROM tickets WHERE ticket_id = $1 FOR UPDATE;
      `, [ticketId]);

      if (current.rows.length === 0) {
        reply.code(404).send({ error: 'Ticket not found' });
        await client.query('ROLLBACK');
        return;
      }

      const currentVersion = current.rows[0].version;
      const newVersion = currentVersion + 1;

      // Build dynamic update query
      const updateFields = [];
      const params = [newVersion, ticketId, currentVersion];
      let paramIdx = 4;

      if (updates.status) {
        updateFields.push(`status = $${paramIdx}`);
        params.push(updates.status);
        paramIdx++;
      }
      if (updates.priority !== undefined) {
        updateFields.push(`priority = $${paramIdx}`);
        params.push(updates.priority);
        paramIdx++;
      }
      if (updates.assignedAgentId) {
        updateFields.push(`assigned_agent_id = $${paramIdx}`);
        params.push(updates.assignedAgentId);
        paramIdx++;
        // Update agent's active ticket count
        await client.query(`
          UPDATE agents SET active_tickets = active_tickets + 1 WHERE user_id = $1;
        `, [updates.assignedAgentId]);
      }
      if (updates.notes) {
        updateFields.push(`notes = $${paramIdx}`);
        params.push(updates.notes);
        paramIdx++;
      }

      updateFields.push(`updated_at = NOW()`);

      const updateQuery = `
        UPDATE tickets
        SET ${updateFields.join(', ')}
        WHERE ticket_id = $2 AND version = $3
        RETURNING ticket_id, status, priority, assigned_agent_id;
      `;

      const updateResult = await client.query(updateQuery, params);
      if (updateResult.rows.length === 0) {
        reply.code(409).send({ error: 'Ticket was updated by another agent, please refresh' });
        await client.query('ROLLBACK');
        return;
      }

      await client.query('COMMIT');

      // Invalidate cache for all agents (simplified, in production use more granular invalidation)
      const keys = await redis.keys('agent:*:tickets:*');
      if (keys.length > 0) {
        await redis.del(keys);
      }

      reply.send(updateResult.rows[0]);
    } catch (err) {
      await client.query('ROLLBACK');
      throw err;
    } finally {
      client.release();
    }
  } catch (err) {
    if (err instanceof z.ZodError) {
      reply.code(400).send({ error: 'Validation failed', details: err.errors });
      return;
    }
    fastify.log.error('Ticket update error:', err);
    reply.code(500).send({ error: 'Failed to update ticket' });
  }
});

// Update agent availability status
fastify.patch('/api/agent/status', async (req, reply) => {
  const agentId = req.agent.user_id;
  const { isAvailable } = req.body;

  if (typeof isAvailable !== 'boolean') {
    reply.code(400).send({ error: 'isAvailable must be a boolean' });
    return;
  }

  try {
    const client = await pgPool.connect();
    await client.query(`
      UPDATE agents SET is_available = $1 WHERE user_id = $2;
    `, [isAvailable, agentId]);
    client.release();

    // Update agent cache
    await redis.setex(`agent:${agentId}`, 300, JSON.stringify({ ...req.agent, is_available: isAvailable }));

    reply.send({ status: 'success', isAvailable });
  } catch (err) {
    fastify.log.error('Agent status update error:', err);
    reply.code(500).send({ error: 'Failed to update status' });
  }
});

// Start server
fastify.listen({ port: 3001, host: '0.0.0.0' }, (err) => {
  if (err) {
    fastify.log.error(err);
    process.exit(1);
  }
});

// Initialize Redis Streams consumer group
await redis.xgroup('ticket:ingestion:stream', 'triage-workers', '$', 'MKSTREAM');

// Optimistic locking update query
const updateQuery = `
  UPDATE tickets
  SET status = $1, version = version + 1
  WHERE ticket_id = $2 AND version = $3
  RETURNING *;
`;

// Cache ticket list with 30s TTL
await redis.setex(`agent:${agentId}:tickets:all`, 30, JSON.stringify(ticketRows));