template-app/convex/relay.ts

// SPDX-License-Identifier: LicenseRef-ConusAI-Template-1.0
// The outbox relay (spec §5.4: outbox events are scheduled internal mutations). A write commits
// its rows and its events in one transaction; this cron-driven pass delivers them to the
// modules' subscribers, at least once. A suspended organization's events are deferred, not
// failed (AT-D19), exactly as the pg-boss relay does on the postgres profile.

import { Container } from "@conusai-kit/core"
import { systemActor } from "@conusai-kit/core/worker"
import { DATA_SOURCE, JOB_QUEUE, OBJECT_STORE, SYSTEM_DATA, toInstant } from "@conusai-kit/data"
import { type ConvexWriter, createConvexDataSource, createConvexJobQueue } from "@conusai-kit/data-convex"
import { executorFor } from "@conusai-kit/data-convex/runtime"
import { FILES_AUDIT, FILES_REGISTRY, FILES_SCOPE, FILES_USAGE } from "@conusai-kit/files"
import { PROPOSALS_AUDIT, PROPOSALS_SCOPE } from "@conusai-kit/proposals"
import { organizations } from "@conusai-kit/tenancy"
import { Audit, defineAuditEvents } from "@conusai-sdk/convex-audit"
import { makeFunctionReference } from "convex/server"
import { components, internal } from "./_generated/api"
import { internalMutation } from "./_generated/server"
import { convexEntities, registry, unavailableObjectStore } from "./lib/runtime"

const auditEvents = defineAuditEvents(
  Object.fromEntries(Object.keys(registry.audit).map((action) => [action, { target: "organization" }])) as never,
)
const audit = new Audit(components.audit, {
  events: auditEvents,
  credentialsProvider: internal.auditCredentials.get,
})
const requiredEnv = (name: string): string => {
  const value = process.env[name]?.trim()
  if (!value) throw new Error(`Convex deployment is missing ${name}`)
  return value
}

/** The next look at a deferred record: one minute doubling per deferral, capped at fifteen. */
export function relayDeferralDelayMs(deferrals: number): number {
  return Math.min(900_000, 60_000 * 2 ** Math.min(deferrals, 30))
}

export const relayPass = internalMutation({
  args: {},
  handler: async (ctx) => {
    const data = createConvexDataSource({ entities: convexEntities, executor: executorFor(ctx) })
    const pending = await data.outbox.pending(50)
    let failureCount = 0
    for (const record of pending) {
      const org = await data.global(organizations).get(record.orgId)
      if (org?.status === "suspended") {
        // Held back without spending an attempt: a suspension is not a delivery failure.
        await data.outbox.defer(
          record.id,
          toInstant(Date.now() + relayDeferralDelayMs(record.deferrals)),
          "the organization is suspended",
        )
        continue
      }
      const entry = registry.events[record.name]
      let failed: string | undefined
      for (const subscription of entry?.subscribers ?? []) {
        const usageEvents: Array<Record<string, unknown>> = []
        try {
          const container = new Container()
          container.value(DATA_SOURCE, data)
          container.value(SYSTEM_DATA, data)
          container.value(OBJECT_STORE, unavailableObjectStore)
          container.value(
            JOB_QUEUE,
            createConvexJobQueue({
              db: ctx.db as unknown as ConvexWriter,
              schedule: ({ delayMs, jobId, inline }) =>
                ctx.scheduler
                  .runAfter(delayMs, makeFunctionReference<"mutation">("conusaiJobs:run"), {
                    jobId,
                    ...(inline ? { inline } : {}),
                  })
                  .then(() => undefined),
            }),
          )
          const deliveryScope = {
            level: "org",
            orgId: record.orgId,
            actor: systemActor("outbox.relay"),
          } as const
          container.value(FILES_SCOPE, deliveryScope)
          container.value(PROPOSALS_SCOPE, deliveryScope)
          container.value(FILES_REGISTRY, registry)
          container.value(FILES_AUDIT, {
            log: async (action: string, fields: Record<string, unknown>) => {
              await audit.log(ctx, action as never, fields as never)
            },
          })
          container.value(PROPOSALS_AUDIT, {
            log: async (action: string, fields: Record<string, unknown>) => {
              await audit.log(ctx, action as never, fields as never)
            },
          })
          container.value(FILES_USAGE, {
            track: (event: Record<string, unknown>) => {
              usageEvents.push(event)
            },
            flush: async () => {
              for (const event of usageEvents.splice(0)) {
                await ctx.scheduler.runAfter(0, makeFunctionReference<"action">("usage:deliver"), {
                  attempt: 0,
                  event,
                  orgSlug: requiredEnv("CONUSAI_ORG_SLUG"),
                })
              }
            },
          })
          await subscription.handler(
            { name: record.name, payload: record.payload },
            { container, correlationId: record.id },
          )
        } catch (error) {
          failed = error instanceof Error ? error.message : String(error)
        }
      }
      if (failed === undefined) await data.outbox.markPublished([record.id])
      else {
        failureCount += 1
        await data.outbox.markFailed(record.id, failed)
      }
    }
    return { delivered: pending.length, failed: failureCount }
  },
})