// `sync.pull` — EXECUTOR (server-only, default export).
//
// The heart of the template: durable `ctx.kv` used for the two kinds of state
// you CAN'T recompute in an external sync.
//
//   1. the CURSOR  — a watermark past everything already pulled. Lose it and
//      you re-process (or skip) events. Read with getOrElse, advanced with set.
//   2. the MARKERS — one per external id, so a redelivered event isn't ingested
//      twice. Written with a TTL (a bounded idempotency window). has() checks
//      one without deserializing.
//
// The event ROWS themselves go to `ctx.store` — they're re-fetchable from the
// source, so a relational table is their home (query/join/report on them).

import type { AppContext } from '@voltro/runtime'

// Stand-in for a real upstream (a webhook backlog, a 3rd-party API page).
// Deterministic + pure so the template boots with zero infra and no network —
// event N has externalId `ext-N`. A real handler would `fetch()` here instead.
const fetchSince = (sequence: number, limit: number) =>
  Array.from({ length: limit }, (_unused, i) => {
    const seq = sequence + i + 1
    return {
      externalId: `ext-${seq}`,
      kind:       seq % 3 === 0 ? 'updated' : 'created',
      payload:    JSON.stringify({ seq }),
      sequence:   seq,
    }
  })

// Idempotency window: how long a "seen" marker lives. Long enough to dedupe
// redelivery, not forever — a bounded lifetime on an otherwise-durable store.
const MARKER_TTL_MS = 24 * 60 * 60_000 // 24h

const execute = async (input: { limit?: number }, ctx: AppContext) => {
  // KV keys are app-global — fold the tenant in so two tenants never collide.
  const tenantId = ctx.request.subject.tenantId ?? 'anon'
  const cursorKey = `sync:${tenantId}:cursor`
  const limit = Math.min(input.limit ?? 5, 50)

  // Durable cursor — getOrElse writes the default (0) ONLY on a genuine
  // first-run miss; on every later call it returns the stored watermark.
  const cursor = await ctx.kv.getOrElse<number>(cursorKey, () => 0)

  let pulled = 0
  let skipped = 0
  let highest = cursor

  for (const evt of fetchSince(cursor, limit)) {
    const markerKey = `sync:${tenantId}:seen:${evt.externalId}`
    if (await ctx.kv.has(markerKey)) {
      // Already ingested within the TTL window — the source redelivered it.
      skipped++
    } else {
      await ctx.store.insert('synced_events', {
        externalId: evt.externalId,
        kind:       evt.kind,
        payload:    evt.payload,
        sequence:   evt.sequence,
        tenantId,
      })
      // Mark seen, with a TTL. Value carries the sequence for traceability.
      await ctx.kv.set(markerKey, evt.sequence, { ttlMs: MARKER_TTL_MS })
      pulled++
    }
    highest = Math.max(highest, evt.sequence)
  }

  // Advance the durable watermark past everything we just saw.
  await ctx.kv.set(cursorKey, highest)

  return { pulled, skipped, cursor: highest }
}

export default execute