Keep a copy in sync

/**
 * Reference receiver for the Totem changes feed. No dependencies: it needs
 * only `fetch`. Compile it with your TypeScript build, or run it directly on
 * Node.js 22.18 or later.
 *
 * It keeps a copy of every team a key reads, safely across crashes:
 * - Run it on your server only. The key is a secret and must never reach a browser.
 * - Run one `syncKey` per key at a time, every 1 to 5 minutes.
 * - Persist the `SyncStore` in your own database. `MemorySyncStore`
 *   (memoryStore.ts) loses everything on restart and is for tests only.
 * - Each page is applied before its cursor is saved, so a crash in between
 *   replays the page, and the version check makes the replay harmless.
 * - One team's trouble never stops the others: a team that cannot be read
 *   right now keeps its copy and its cursor, and the next run tries again.
 * - It stops instead of looping if the API ever repeats a cursor or sends
 *   none, before saving it, so a fault on either side cannot spend the key's
 *   daily quota.
 */

export interface FeedItem {
  kind: string
  id: string
  version: number
  object?: Record<string, unknown>
  removed?: { reason: string }
}

export interface FeedPage {
  object: 'list'
  data: FeedItem[]
  has_more: boolean
  next_cursor: string
}

interface ListPage {
  data: unknown[]
  has_more: boolean
  next_cursor: string | null
}

/** Your persistence. Every method must be durable before its promise resolves. */
export interface SyncStore {
  teamIds(): Promise<string[]>
  ensureTeam(teamId: string): Promise<void>
  cursor(teamId: string): Promise<string | null>
  /** The version you hold for this record, removals included; 0 when you hold none. */
  version(teamId: string, kind: string, id: string): Promise<number>
  setContactPermissions(teamId: string, leadId: string, permissions: Record<string, unknown>): Promise<void>
  upsert(teamId: string, item: FeedItem): Promise<void>
  /**
   * Delete the record but keep its version (a tombstone). Removing a lead also
   * deletes its attribution, response and activities.
   */
  remove(teamId: string, item: FeedItem): Promise<void>
  saveCursor(teamId: string, cursor: string): Promise<void>
  /** Start a full resync: forget the team's versions and cursor, keep its records, clear the seen set. */
  startResync(teamId: string): Promise<void>
  /** Remember that the feed still delivers this record. Must survive a restart until the resync finishes. */
  markSeen(teamId: string, item: FeedItem): Promise<void>
  /** End a full resync: delete the team's records the feed did not deliver. A no-op when no resync is running. */
  finishResync(teamId: string): Promise<void>
  /** Delete everything you hold for the team, cursor included. */
  dropTeam(teamId: string): Promise<void>
  dictionariesAt(teamId: string): Promise<number | null>
  saveDictionaries(teamId: string, dictionaries: Record<string, unknown[]>, at: number): Promise<void>
  /**
   * True when the item names a board, stage, member or service that is neither
   * in your dictionaries nor remembered as unresolved (`rememberUnresolved`).
   */
  hasUnknownReferences(teamId: string, item: FeedItem): Promise<boolean>
  /**
   * Called for an item that still names unknown ids right after the
   * dictionaries were refreshed. A record can keep naming a member who left
   * the team, or a board, stage or service that was deleted: remember those
   * ids so they never cause another refresh. The daily refresh still picks
   * up any that come back.
   */
  rememberUnresolved(teamId: string, item: FeedItem): Promise<void>
}

export interface SyncOptions {
  /** A secret key (`tsk_live_…`). */
  key: string
  store: SyncStore
  fetch?: typeof globalThis.fetch
  baseUrl?: string
  sleep?: (ms: number) => Promise<void>
  now?: () => number
  /** How many times one request is retried after a `429` or a `5xx`. Default 3. */
  maxRetries?: number
  /**
   * The longest this process waits in one go, in milliseconds. Default 60 000.
   * A longer `Retry-After` (the daily quota resets at 00:00 UTC) ends that
   * team's run with `retryAt` instead of keeping the process asleep.
   */
  maxWaitMs?: number
  /**
   * The most feed pages one team reads in one run. No limit by default. A
   * team's initial sync can take hours: with a limit, it stops with `partial`
   * after that many pages, the key's other teams are synced, and the next run
   * continues from the saved cursor.
   */
  maxPagesPerTeam?: number
}

/** What happened to one team in one `syncKey` run. */
export type TeamOutcome =
  /** Caught up; the cursor is saved. */
  | { team: string, status: 'synced' }
  /** Read `maxPagesPerTeam` pages and not caught up yet; the cursor is saved and the next run continues. */
  | { team: string, status: 'partial' }
  /** The team left the key: its copy and cursor are deleted. */
  | { team: string, status: 'dropped' }
  /** Not readable right now (`403`, `capture_not_ready`, unconfirmed `team_not_accessible`); copy and cursor kept. */
  | { team: string, status: 'skipped', code: string }
  /**
   * Asked to wait longer than `maxWaitMs`, or still failing after `maxRetries`;
   * copy and cursor kept. Do not call before `retryAt` (epoch milliseconds).
   */
  | { team: string, status: 'retry_later', code: string, retryAt: number }

/** An error answer from the API. `retryAt` (epoch milliseconds) is set when asking again later can succeed. */
export class SyncHttpError extends Error {
  readonly status: number
  readonly code: string
  readonly retryAt: number | null

  constructor(status: number, code: string, retryAt: number | null = null) {
    super(`Totem API answered ${status} ${code}`)
    this.name = 'SyncHttpError'
    this.status = status
    this.code = code
    this.retryAt = retryAt
  }
}

const DEFAULT_BASE_URL = 'https://api.totemstudio.ai/v1'
const DAY_MS = 86_400_000

/**
 * The API always moves a cursor forward. Throws when a page says there is
 * more but repeats a cursor this loop already used, or sends none: following
 * it would read the same page forever.
 */
function nextCursor(seen: Set<string>, cursor: string | null): string {
  if (!cursor || seen.has(cursor)) throw new Error('Totem API sent a missing or repeated cursor; stopping so the loop cannot spend the key\'s quota')
  seen.add(cursor)
  return cursor
}

/** `Retry-After` is always whole seconds. Null when the header is missing or unreadable. */
function retryAfterMs(response: Response): number | null {
  const header = response.headers.get('Retry-After')
  if (header === null || header.trim() === '') return null
  const seconds = Number(header)
  return Number.isFinite(seconds) && seconds >= 0 ? seconds * 1000 : null
}

/** One GET. A `429` or `5xx` is retried while the wait fits within `maxWaitMs`. */
async function get<T>(options: SyncOptions, path: string, team?: string): Promise<T> {
  const fetcher = options.fetch ?? globalThis.fetch
  const sleep = options.sleep ?? (ms => new Promise<void>(resolve => setTimeout(resolve, ms)))
  const now = options.now ?? Date.now
  const maxRetries = options.maxRetries ?? 3
  const maxWaitMs = options.maxWaitMs ?? 60_000

  for (let attempt = 0; ; attempt++) {
    const response = await fetcher(`${options.baseUrl ?? DEFAULT_BASE_URL}${path}`, {
      headers: { Authorization: `Bearer ${options.key}`, ...(team ? { 'Totem-Team': team } : {}) },
    })
    if (response.ok) return await response.json() as T

    // Errors are application/problem+json. The firewall in front of the API may answer without a body.
    const problem = await response.json().catch(() => ({})) as { code?: string }
    const code = problem.code ?? `http_${response.status}`
    // Only a 429 or a 5xx is worth sending again within this run; the caller decides about the rest.
    if (response.status !== 429 && response.status < 500) throw new SyncHttpError(response.status, code)

    const waitMs = retryAfterMs(response) ?? 1000 * 2 ** attempt
    // Too long to wait here (the daily quota, say), or out of retries: report when to come back.
    if (waitMs > maxWaitMs || attempt >= maxRetries) throw new SyncHttpError(response.status, code, now() + waitMs)
    await sleep(waitMs)
  }
}

/** Every team `GET /v1/teams` lists. It lists restricted teams too, so a missing team has left the key. */
async function listedTeams(options: SyncOptions): Promise<string[]> {
  const page = await get<{ data: Array<{ id: string }> }>(options, '/teams')
  return page.data.map(team => team.id)
}

/**
 * Members and bookable services are not in the feed: read them in full when a
 * team starts, when an item names an id you do not know, and once a day.
 * Boards do travel in the feed; they are read here as well because a lead can
 * name a new stage before that board's change reaches you.
 */
async function refreshDictionaries(options: SyncOptions, teamId: string): Promise<void> {
  const boards: unknown[] = []
  const seen = new Set<string>()
  let cursor: string | null = null
  do {
    const query: string = cursor ? `&cursor=${encodeURIComponent(cursor)}` : ''
    const page: ListPage = await get<ListPage>(options, `/boards?limit=100${query}`, teamId)
    boards.push(...page.data)
    cursor = page.has_more ? nextCursor(seen, page.next_cursor) : null
  } while (cursor)
  // These two arrive whole, in one response, and take no `limit` or `cursor`.
  const members = await get<ListPage>(options, '/members', teamId)
  const services = await get<ListPage>(options, '/booking_services', teamId)
  const now = options.now ?? Date.now
  const dictionaries = { boards, members: members.data, booking_services: services.data }
  await options.store.saveDictionaries(teamId, dictionaries, now())
}

/** Applies one page. Every item is checked against the version you hold, removals included. */
async function applyPage(options: SyncOptions, teamId: string, page: FeedPage): Promise<void> {
  const { store } = options
  let refreshed = false
  for (const item of page.data) {
    // Apply only what is newer than your copy. A removal keeps its version as a
    // tombstone, so an older copy replayed later cannot bring the record back.
    if (item.version > await store.version(teamId, item.kind, item.id)) {
      if (await store.hasUnknownReferences(teamId, item)) {
        // One refresh per page at most. What it does not resolve (a member who
        // left, a deleted stage) is remembered, so it never asks again.
        if (!refreshed) {
          await refreshDictionaries(options, teamId)
          refreshed = true
        }
        await store.rememberUnresolved(teamId, item)
      }
      if (item.removed) {
        await store.remove(teamId, item)
      }
      else {
        // Commit contact permissions before anything downstream can act on the lead.
        const permissions = item.kind === 'lead' ? item.object?.contact_permissions : undefined
        if (permissions && typeof permissions === 'object') {
          await store.setContactPermissions(teamId, item.id, permissions as Record<string, unknown>)
        }
        await store.upsert(teamId, item)
      }
    }
    // During a full resync, remember every record the feed still delivers.
    await store.markSeen(teamId, item)
  }
}

/**
 * Brings one team's copy up to date: reads the feed from the saved cursor
 * until it has caught up (`synced`), or until it has read `maxPagesPerTeam`
 * pages (`partial`). Throws `SyncHttpError` when the API refuses; the saved
 * cursor is untouched, so the next run resumes from the same place.
 */
export async function syncTeam(options: SyncOptions & { teamId: string }): Promise<'synced' | 'partial'> {
  const { store, teamId } = options
  const now = options.now ?? Date.now
  await store.ensureTeam(teamId)
  const dictionariesAt = await store.dictionariesAt(teamId)
  if (dictionariesAt === null || now() - dictionariesAt >= DAY_MS) await refreshDictionaries(options, teamId)

  let resyncing = false
  let pages = 0
  let seen = new Set<string>()
  for (;;) {
    const cursor = await store.cursor(teamId)
    const query = cursor ? `&cursor=${encodeURIComponent(cursor)}` : ''
    let page: FeedPage
    try {
      page = await get<FeedPage>(options, `/sync/changes?limit=100${query}`, teamId)
    }
    catch (error) {
      // The cursor expired (410) or the feed cannot continue from it (409 resync_required):
      // run a full resync, once per run.
      const restart = error instanceof SyncHttpError && (error.status === 410 || error.code === 'resync_required')
      if (restart && !resyncing) {
        await store.startResync(teamId)
        resyncing = true
        seen = new Set()
        continue
      }
      throw error
    }
    await applyPage(options, teamId, page)
    // Check the cursor before saving it, so a missing or repeated one never
    // replaces the last good cursor. A last page only has to send one.
    if (cursor) seen.add(cursor)
    if (page.has_more || !page.next_cursor) nextCursor(seen, page.next_cursor)
    // Save the cursor only after the page is applied, and save it for an empty
    // page too: that keeps the cursor of a quiet team from expiring.
    await store.saveCursor(teamId, page.next_cursor)
    if (!page.has_more) {
      // After a full resync, delete what the feed no longer delivers.
      await store.finishResync(teamId)
      return 'synced'
    }
    pages++
    // Leave the rest for the next run, so the key's other teams are not kept waiting.
    if (options.maxPagesPerTeam !== undefined && pages >= options.maxPagesPerTeam) return 'partial'
  }
}

/**
 * What one team's error means for this run. Every outcome but `dropped` keeps
 * the team's copy and cursor. Any other error (a `401`, a `400`, an unexpected
 * `404`, a feed that still asks for a resync after one) concerns the whole key
 * or a bug, so it is rethrown and stops the run.
 */
async function outcomeOf(options: SyncOptions, team: string, error: SyncHttpError): Promise<TeamOutcome> {
  // A 429 or 5xx that asked for a longer wait than this run allows (or kept failing).
  if (error.retryAt !== null) return { team, status: 'retry_later', code: error.code, retryAt: error.retryAt }
  // Restricted, sector-restricted or switched off: keep everything and skip the team.
  if (error.status === 403) return { team, status: 'skipped', code: error.code }
  // Totem is still preparing this team's data (once per team, ever): a later run reads it.
  if (error.code === 'capture_not_ready') return { team, status: 'skipped', code: error.code }
  // The team may have left the key. Only `/teams` can confirm it; a team it still lists keeps its copy.
  if (error.code === 'team_not_accessible') {
    if ((await listedTeams(options)).includes(team)) return { team, status: 'skipped', code: error.code }
    await options.store.dropTeam(team)
    return { team, status: 'dropped' }
  }
  throw error
}

/**
 * Syncs every team the key reads and reports what happened to each. Re-reads
 * `/teams` on every run, so a new team starts its initial sync and a team that
 * left the key is deleted. Once a `429` (the daily quota, say) sends the key
 * away for longer than `maxWaitMs`, the teams not reached yet are reported
 * `retry_later` for the same time, without a request.
 *
 * Throws when the key itself is refused, when `/teams` cannot be read, when a
 * team's answer points to a bug rather than to the team (a `400`, say), or
 * when the network fails. No cursor is lost. When the error carries `retryAt`
 * (the daily quota is spent, for example), schedule the next run for then.
 */
export async function syncKey(options: SyncOptions): Promise<TeamOutcome[]> {
  const outcomes: TeamOutcome[] = []
  const listed = await listedTeams(options)
  for (const team of await options.store.teamIds()) {
    if (listed.includes(team)) continue
    await options.store.dropTeam(team)
    outcomes.push({ team, status: 'dropped' })
  }
  // Set when a `429` sends the key away for longer than this run waits.
  let keyWait: { code: string, retryAt: number } | null = null
  for (const team of listed) {
    if (keyWait) {
      outcomes.push({ team, status: 'retry_later', ...keyWait })
      continue
    }
    try {
      outcomes.push({ team, status: await syncTeam({ ...options, teamId: team }) })
    }
    catch (error) {
      if (!(error instanceof SyncHttpError)) throw error
      const outcome = await outcomeOf(options, team, error)
      outcomes.push(outcome)
      // A `429` concerns the key, never one team: the daily quota and the rate
      // limit are both the key's. The teams not reached yet would be refused
      // too, so they wait for the same moment without spending a request.
      if (error.status === 429 && outcome.status === 'retry_later') keyWait = { code: outcome.code, retryAt: outcome.retryAt }
    }
  }
  return outcomes
}

/**
 * An in-memory `SyncStore`, for tests and for reading alongside syncTeam.ts.
 * It loses everything on restart: in production, implement every method on
 * your database, durably, and commit a page and its cursor together when you can.
 */
import type { FeedItem, SyncStore } from './syncTeam.js'

/** One key per record: a lead and its two facets share an id but not a kind. */
const keyOf = (item: Pick<FeedItem, 'kind' | 'id'>) => `${item.kind}/${item.id}`

/**
 * Every board, stage, service and member id an item names: a lead's `board`,
 * `stage` and `assignees`, a conversation's `assignee`, a booking's `service`,
 * and an activity's `stage_snapshot`, `from` and `to`: a stage with its board,
 * or a list of members. A status change's `from` and `to` are statuses.
 */
function referencedIds(item: FeedItem): string[] {
  const object = item.object ?? {}
  const ids: unknown[] = [object.board, object.stage, object.service, object.assignee]
  if (Array.isArray(object.assignees)) ids.push(...object.assignees)
  if (item.kind === 'activity') {
    for (const side of [object.from, object.to, object.stage_snapshot]) {
      if (Array.isArray(side)) {
        ids.push(...side)
      }
      else if (side && typeof side === 'object') {
        const { stage, board } = side as { stage?: unknown, board?: unknown }
        ids.push(stage, board)
      }
    }
  }
  return ids.filter((value): value is string => typeof value === 'string')
}

export class MemorySyncStore implements SyncStore {
  /** team → record key → the latest item applied. */
  rows = new Map<string, Map<string, FeedItem>>()
  cursors = new Map<string, string>()
  /** team → record key → version held, removals (tombstones) included. */
  versions = new Map<string, Map<string, number>>()
  /** `team/lead` → the lead's contact permissions. */
  permissions = new Map<string, Record<string, unknown>>()
  dictionaries = new Map<string, Record<string, unknown[]>>()
  dictionaryTimes = new Map<string, number>()
  /** team → keys the feed delivered during a running full resync. */
  recovery = new Map<string, Set<string>>()
  /** `team/lead` of every removed lead, so a facet or activity replayed later is not stored again. */
  removedLeads = new Set<string>()
  /** team → ids still unknown right after a dictionary refresh: departed members, deleted boards, stages or services. */
  unresolved = new Map<string, Set<string>>()

  async teamIds() {
    return [...this.rows.keys()]
  }

  async ensureTeam(team: string) {
    if (!this.rows.has(team)) this.rows.set(team, new Map())
    if (!this.versions.has(team)) this.versions.set(team, new Map())
  }

  async cursor(team: string) {
    return this.cursors.get(team) ?? null
  }

  async version(team: string, kind: string, id: string) {
    return this.versions.get(team)?.get(`${kind}/${id}`) ?? 0
  }

  async setContactPermissions(team: string, lead: string, permissions: Record<string, unknown>) {
    this.permissions.set(`${team}/${lead}`, structuredClone(permissions))
  }

  async upsert(team: string, item: FeedItem) {
    await this.ensureTeam(team)
    this.versions.get(team)!.set(keyOf(item), item.version)
    // A facet or an activity of a removed lead keeps its version but is not stored.
    const lead = item.kind === 'lead_attribution' || item.kind === 'lead_response' ? item.id : item.object?.lead
    if (typeof lead === 'string' && this.removedLeads.has(`${team}/${lead}`)) return
    this.rows.get(team)!.set(keyOf(item), structuredClone(item))
  }

  async remove(team: string, item: FeedItem) {
    await this.ensureTeam(team)
    // The tombstone: the version stays, so an older copy delivered later is ignored.
    this.versions.get(team)!.set(keyOf(item), item.version)
    const rows = this.rows.get(team)!
    rows.delete(keyOf(item))
    if (item.kind !== 'lead') return
    // A removed lead takes its attribution, response, activities and permissions with it.
    this.removedLeads.add(`${team}/${item.id}`)
    this.permissions.delete(`${team}/${item.id}`)
    for (const [key, row] of rows) {
      const facet = (row.kind === 'lead_attribution' || row.kind === 'lead_response') && row.id === item.id
      const activity = row.kind === 'activity' && row.object?.lead === item.id
      if (facet || activity) rows.delete(key)
    }
  }

  async saveCursor(team: string, cursor: string) {
    this.cursors.set(team, cursor)
  }

  async startResync(team: string) {
    this.versions.set(team, new Map())
    this.cursors.delete(team)
    this.recovery.set(team, new Set())
    this.unresolved.delete(team)
    // A restore can reissue ids and versions, so the resync starts from a clean slate.
    for (const key of this.removedLeads) if (key.startsWith(`${team}/`)) this.removedLeads.delete(key)
  }

  async markSeen(team: string, item: FeedItem) {
    this.recovery.get(team)?.add(keyOf(item))
  }

  async finishResync(team: string) {
    const seen = this.recovery.get(team)
    if (!seen) return
    const rows = this.rows.get(team) ?? new Map<string, FeedItem>()
    for (const [key, row] of rows) {
      if (seen.has(key)) continue
      rows.delete(key)
      if (row.kind === 'lead') this.permissions.delete(`${team}/${row.id}`)
    }
    this.recovery.delete(team)
  }

  async dropTeam(team: string) {
    this.rows.delete(team)
    this.cursors.delete(team)
    this.versions.delete(team)
    this.recovery.delete(team)
    this.dictionaries.delete(team)
    this.dictionaryTimes.delete(team)
    this.unresolved.delete(team)
    for (const key of this.permissions.keys()) if (key.startsWith(`${team}/`)) this.permissions.delete(key)
    for (const key of this.removedLeads) if (key.startsWith(`${team}/`)) this.removedLeads.delete(key)
  }

  async dictionariesAt(team: string) {
    return this.dictionaryTimes.get(team) ?? null
  }

  async saveDictionaries(team: string, data: Record<string, unknown[]>, at: number) {
    this.dictionaries.set(team, structuredClone(data))
    this.dictionaryTimes.set(team, at)
  }

  async hasUnknownReferences(team: string, item: FeedItem) {
    const unresolved = this.unresolved.get(team)
    return this.unknownIds(team, item).some(id => !unresolved?.has(id))
  }

  async rememberUnresolved(team: string, item: FeedItem) {
    const ids = this.unknownIds(team, item)
    if (!ids.length) return
    const unresolved = this.unresolved.get(team) ?? new Set<string>()
    for (const id of ids) unresolved.add(id)
    this.unresolved.set(team, unresolved)
  }

  /** The board, stage, service and member ids the item names that the dictionaries do not hold. */
  private unknownIds(team: string, item: FeedItem): string[] {
    if (!item.object) return []
    const known = new Set<string>()
    for (const entry of Object.values(this.dictionaries.get(team) ?? {}).flat()) {
      const object = entry as { id: string, stages?: Array<{ id: string }> }
      known.add(object.id)
      for (const stage of object.stages ?? []) known.add(stage.id)
    }
    return referencedIds(item).filter(id => !known.has(id))
  }
}