template-app/convex/relay.ts

// SPDX-License-Identifier: LicenseRef-ConusAI-Template-1.0
// The outbox relay (spec §5.4). A write commits its rows and its events in one transaction; the
// cron's action (`relayPass`) delivers them to the modules' subscribers through the kit's relay
// steps, each its own mutation: `claimEvents` counts an attempt and leases the records,
// `deliverEvent` runs one subscriber with its delivered mark, and `settleEvent` publishes the
// record or records the failure. A subscriber that throws takes its writes with it; a redelivery
// skips the subscribers already delivered; one record's failure never undoes another's. 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,
  claimConvexRelayBatch,
  createConvexDataSource,
  createConvexJobQueue,
  deliverConvexEvent,
  runConvexRelayPass,
  settleConvexRelayRecord,
} 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 { NOTIFICATIONS_REPOS } from "@conusai-kit/notifications"
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 { v } from "convex/values"
import { components, internal } from "./_generated/api"
import { internalAction, internalMutation } from "./_generated/server"
import { convexEntities, convexLive, 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))
}

const CLAIM_EVENTS = makeFunctionReference<"mutation">("relay:claimEvents")
const DELIVER_EVENT = makeFunctionReference<"mutation">("relay:deliverEvent")
const SETTLE_EVENT = makeFunctionReference<"mutation">("relay:settleEvent")

/** The cron's pass: every step below is its own transaction. */
export const relayPass = internalAction({
  args: {},
  handler: async (ctx) =>
    runConvexRelayPass({
      claim: () => ctx.runMutation(CLAIM_EVENTS, {}),
      deliver: (delivery) => ctx.runMutation(DELIVER_EVENT, delivery),
      settle: (settlement) => ctx.runMutation(SETTLE_EVENT, settlement),
    }),
})

/** Counts an attempt on each record this pass takes, before any of its deliveries can fail. */
export const claimEvents = internalMutation({
  args: {},
  handler: async (ctx) => {
    const data = createConvexDataSource({ entities: convexEntities, live: convexLive, executor: executorFor(ctx) })
    return claimConvexRelayBatch(ctx.db as unknown as ConvexWriter, {
      registry,
      // Held back without spending an attempt: a suspension is not a delivery failure.
      hold: async (record) => {
        const org = await data.global(organizations).get(record.orgId)
        return org?.status === "suspended"
          ? {
              until: toInstant(Date.now() + relayDeferralDelayMs(record.deferrals)),
              reason: "the organization is suspended",
            }
          : undefined
      },
    })
  },
})

/** Publishes a record every subscriber has, or records why one did not (stuck after the last attempt). */
export const settleEvent = internalMutation({
  args: { recordId: v.string(), error: v.optional(v.string()) },
  handler: async (ctx, args) => settleConvexRelayRecord(ctx.db as unknown as ConvexWriter, args, { registry }),
})

/** One (record, subscriber) delivery, in its own transaction, with the subscriber's ports. */
export const deliverEvent = internalMutation({
  args: { recordId: v.string(), subscriber: v.string() },
  handler: async (ctx, args) => {
    const data = createConvexDataSource({ entities: convexEntities, live: convexLive, executor: executorFor(ctx) })
    return deliverConvexEvent(ctx.db as unknown as ConvexWriter, args, {
      registry,
      handle: async (record, payload, subscription) => {
        const usageEvents: Array<Record<string, unknown>> = []
        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(NOTIFICATIONS_REPOS, data.forScope(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 }, { container, correlationId: record.id })
      },
    })
  },
})