API sync — ticketing

import { type ExuluContextSource, type ExuluItem } from "@exulu/backend";
import { ExuluQueues } from "@exulu/backend";
import { createHash } from "crypto";

// ── Types ─────────────────────────────────────────────────────────────────────

type ApiTicket = {
  id:         string;
  subject:    string;
  body:       string;
  status:     "open" | "pending" | "solved" | "closed";
  priority:   "low" | "normal" | "high" | "urgent";
  submitter:  string;
  created_at: string;
  updated_at: string;
};

type TicketsPage = {
  tickets:       ApiTicket[];
  next_cursor:   string | null;
  end_of_stream: boolean;
};

// ── Rate-limit-aware fetcher ──────────────────────────────────────────────────

const MAX_RETRIES = 3;

async function fetchPage(params: {
  cursor:     string | null;
  limit:      number;
  start_time?: string;
  attempt?:   number;
}): Promise<TicketsPage> {
  const { cursor, limit, start_time, attempt = 0 } = params;

  const url = new URL("https://api.example.com/v2/tickets");
  url.searchParams.set("limit",  String(limit));
  url.searchParams.set("status", "solved");
  if (cursor)     url.searchParams.set("cursor",     cursor);
  if (start_time) url.searchParams.set("start_time", start_time);

  const res = await fetch(url.toString(), {
    headers: { Authorization: `Bearer ${process.env.TICKETING_API_TOKEN}` },
  });

  // ── 429: honour Retry-After with exponential fallback ────────────────────
  if (res.status === 429) {
    if (attempt >= MAX_RETRIES) {
      throw new Error(`Ticketing API rate-limited after ${MAX_RETRIES} retries.`);
    }
    const retryAfter = Number(res.headers.get("Retry-After") ?? 0);
    const backoff    = retryAfter > 0
      ? retryAfter * 1_000
      : Math.min(2 ** attempt * 1_000, 30_000); // 1s, 2s, 4s … capped at 30s

    console.warn(
      `[ticketing_source] 429 received — waiting ${backoff}ms (attempt ${attempt + 1}/${MAX_RETRIES})`,
    );
    await new Promise((r) => setTimeout(r, backoff));
    return fetchPage({ ...params, attempt: attempt + 1 });
  }

  if (!res.ok) {
    throw new Error(`Ticketing API ${res.status}: ${await res.text()}`);
  }

  return res.json() as Promise<TicketsPage>;
}

// ── Full incremental fetch ────────────────────────────────────────────────────

async function fetchAllTickets(startTime?: string): Promise<ApiTicket[]> {
  const all: ApiTicket[] = [];
  let cursor: string | null = null;
  let page = 0;

  do {
    const result = await fetchPage({ cursor, limit: 100, start_time: startTime });
    all.push(...result.tickets);
    cursor = result.next_cursor;
    page++;
    if (result.end_of_stream) break;
  } while (cursor);

  console.log(`[ticketing_source] Fetched ${all.length} tickets across ${page} pages`);
  return all;
}

// ── Item mapping ──────────────────────────────────────────────────────────────

function toExuluItem(ticket: ApiTicket): ExuluItem {
  const content_hash = createHash("sha256")
    .update(ticket.subject + ticket.body + ticket.status + ticket.updated_at)
    .digest("hex");

  return {
    external_id:  ticket.id,
    name:         ticket.subject,
    body:         ticket.body,
    status:       ticket.status,
    priority:     ticket.priority,
    submitter:    ticket.submitter,
    created_at:   new Date(ticket.created_at),
    content_hash,
    rights_mode:  "users",
  };
}

// ── Queue ─────────────────────────────────────────────────────────────────────

export const ticketSyncQueue = ExuluQueues.register(
  "ticket_api_sync",
  { worker: 1, queue: 1 }, // one sync at a time
  1,
  60 * 30,                 // 30-minute timeout
);

// ── Source definition ─────────────────────────────────────────────────────────

export const ticketingApiSource: ExuluContextSource = {
  id:          "ticketing_api_source",
  name:        "Ticketing API source",
  description: "Pulls solved tickets updated since the last run; skips unchanged items via content hash.",

  config: {
    schedule: "0 2 * * *",  // daily at 02:00 UTC
    queue:    ticketSyncQueue.use(),
    retries:  3,
    backoff:  { type: "exponential", delay: 2_000 },
    params: [
      {
        name:        "start_time",
        description: "ISO datetime — fetch tickets updated after this time. Defaults to 25 hours ago.",
      },
      {
        name:        "full_sync",
        description: "Set to 'true' to fetch all tickets regardless of start_time.",
        default:     "false",
      },
    ],
  },

  execute: async ({ start_time: inputStartTime, full_sync: inputFullSync, exuluConfig }) => {
    const fullSync  = inputFullSync === "true";
    const startTime = fullSync
      ? undefined
      : (inputStartTime ?? new Date(Date.now() - 25 * 60 * 60 * 1_000).toISOString());

    console.log("[ticketing_api_source] Starting sync", { startTime, fullSync });

    // ── Fetch from API ────────────────────────────────────────────────────────
    const tickets = await fetchAllTickets(startTime);

    // ── Map to ExuluItem[] with partial-failure logging ───────────────────────
    const items: ExuluItem[] = [];
    for (const ticket of tickets) {
      try {
        items.push(toExuluItem(ticket));
      } catch (err) {
        // Log and continue — one bad ticket should not abort the whole sync
        console.error(`[ticketing_api_source] Failed to map ticket ${ticket.id}:`, err);
      }
    }

    console.log(`[ticketing_api_source] Returning ${items.length} items (${tickets.length - items.length} mapping failures)`);
    return items;
  },
};

import { ticketingApiSource } from "../integrations/ticketing/source";

export const supportTicketsContext = new ExuluContext({
  id: "support_tickets_context",
  // ... fields, embedder, chunker, configuration from the tutorials

  sources: [ticketingApiSource],
});