Architecture

Last updated: 2026-07-31

This document describes the internal architecture of the platform's major subsystems: background jobs, caching, rate limiting, SSE events, audit & operation logging, system configuration, file uploads, the AI agent, permissions (RBAC), notifications, and API tokens. Each section covers the data model, core components, configuration, and caveats for one subsystem.

Architecture: Job System

The platform ships an in-process background job queue built on top of Postgres (for durability) and p-queue (for in-memory concurrency control). Work is defined by Job templates (recurring via cron or manual trigger); templates produce JobInstance rows that are picked up by an in-process scheduler/worker. Instances keep their terminal status in place — there is no separate archive table.

All Job code lives under packages/service/src/lib/queues/. The job queue is in-process — there is no external broker (no Redis/BullMQ). It runs inside the same Node process that serves the Hono API (mounted in the gateway app).

1. Data Model

Defined in packages/service/prisma/schema.prisma.

enum JobStatus    { PENDING | PROCESSING | COMPLETED | FAILED }
enum JobPriority  { CRITICAL | HIGH | NORMAL | LOW | IDLE }

Job (schema.prisma, table job) — the template: defines a group of jobs (recurring or manual-trigger). Templates are mutable.

Field	Purpose
`id`	cuid
`name`	Unique human-friendly identifier (e.g. `"session-sweep"`)
`type`	Handler key, e.g. `"send-notification"`
`payload`	`Json?` — default payload copied into each produced instance
`cronExpression`	`String?` — 5-field cron. `NULL` = manual-trigger only
`enabled`	When `false`, the recurring schedule is paused
`priority` / `maxAttempts` / `timeoutMs`	Defaults inherited by instances
`lastRunAt`	When the last instance was dispatched
`nextRunAt`	Pre-computed next cron occurrence (via `croner`)

JobInstance (table job_instance) — a single execution, produced either by a template (jobId set) or ad-hoc by application code (jobId = null).

Field	Purpose
`id`	cuid
`jobId`	`String?` — FK to the template; `null` for ad-hoc/event-driven jobs
`type`	Handler key (denormalized from template or set directly)
`payload`	`Json` — opaque data passed to the handler
`status`	Current lifecycle state
`priority`	Ordering hint (currently informational)
`attempts` / `maxAttempts`	Retry bookkeeping (default max 3)
`timeoutMs`	Per-instance execution timeout (default 60 000 ms)
`scheduledAt`	When the instance becomes eligible to run
`startedAt` / `completedAt`	Lifecycle timestamps
`result` / `error`	Handler return value / failure message

On template deletion, JobInstance.jobId is set to null (onDelete: SetNull) so execution history is preserved.

2. Core Components

                    ┌─────────────────────────────────────────────────────┐
                    │              JobExecutor (facade)                     │
                    │  start() · enqueue() · subscribe() · stats()          │
                    └──────────────┬─────────────────────────┬─────────────┘
         ┌────────────────────────┘                           │
         ▼                                                    ▼
   ┌─────────────┐   ┌─────────────┐   ┌─────────────┐   ┌──────────────────┐
   │JobScheduler │   │  JobQueue   │   │  JobWorker  │   │JobTemplateScheduler│
   │(instances)  │   │(p-queue, N) │   │(run+retry)  │   │  (cron dispatch)  │
   └─────────────┘   └─────────────┘   └─────────────┘   └──────────────────┘

JobQueue — `lib/queues/job-queue.ts`

A thin wrapper around p-queue that enforces concurrency. A single JobProcessor callback is registered by the executor. Concurrency comes from JOB_CONCURRENCY (default 5).

JobScheduler — `lib/queues/job-scheduler.ts` (instance scheduler)

Decides when a JobInstance enters the in-memory queue.

On start() it resets orphaned PROCESSING rows to PENDING (recoverStuckProcessing), then calls loadExpiredJobs(): atomically claims due PENDING instances in batches of 1000 via claimDueJobs (SELECT … FOR UPDATE SKIP LOCKED), enqueues each claimed instance, and arms a setTimeout for the next future-scheduled instance.
Listens to the job:created event: atomically claims the instance via claimJobById (conditional updateMany on status = PENDING) before enqueueing, so concurrent loadExpiredJobs and job:created paths cannot double-enqueue the same row.
Listens to the job:rescheduled event to re-arm the timer.
Long delays are capped at MAX_TIMER_DURATION_MS (24h) and re-evaluated on fire.

JobTemplateScheduler — `lib/queues/job-template-scheduler.ts` (recurring engine)

Produces JobInstance rows from due templates — this is what makes recurring jobs work.

On start() it calls dispatchDue(): atomically claims due templates via claimDueTemplates (SELECT ... FOR UPDATE SKIP LOCKED), advancing each template's lastRunAt/nextRunAt within the same transaction, then creates a JobInstance for each claimed template (copying type/payload defaults) and emits job:created (→ the instance scheduler enqueues it). The schedule is advanced inside the claim so a transient error during instance creation cannot cause re-dispatch storms.
Skip-semantics: nextRunAt is always computed from "now", so downtime gaps produce one catch-up run on recovery, not a backlog of missed runs.
After dispatch it arms a setTimeout for the next due template (capped at 24h, same pattern as the instance scheduler). The executor's rearmTemplateScheduler() is called after template create/update/delete.

JobWorker — `lib/queues/job-worker.ts`

Processes one instance. This is the heart of the execution model:

Mark PROCESSING, set startedAt, increment attempts.
Look up the handler by job.type in the registry (404-ish error if missing).
Run handler(job) racing against a setTimeout(job.timeoutMs) timeout.
On success → COMPLETED + store result.
On failure:
- If attempts >= maxAttempts → FAILED + store error.
- Otherwise → set back to PENDING with scheduledAt = now + backoff, where backoff = min(5 000ms · 2^(attempts-1), 5min) (exponential, capped).

There is no archiving — the instance row simply retains its terminal status. The job-instance-sweep handler (see §6) prunes COMPLETED and FAILED instances older than 30 days on a daily schedule.

JobHandlerRegistry — `lib/queues/job-handler-registry.ts`

A Record<type, JobHandler> lookup. Handlers are registered at startup (see §5). JobHandler is (job: JobInstance) => Promise<unknown> (job.types.ts).

JobExecutorContext — `lib/queues/job-executor-context.ts`

A tiny typed event emitter for the five lifecycle events: job:created | job:processing | job:completed | job:failed | job:rescheduled.

JobExecutor — `lib/queues/job-executor.ts`

The facade that wires everything together. Public API:

start() — boots both schedulers (instance recovery + template dispatch).
enqueue(job) — emits job:created (instance scheduler decides if it runs now).
rearmTemplateScheduler() — re-arms the template timer after template changes.
subscribe(fn) — fan-out over all five lifecycle events.
getStats() — live queueSize, pending, concurrency from p-queue.

3. Lifecycle of a Job Instance

  create  ──►  PENDING  ──(due)──►  PROCESSING  ──►  COMPLETED
                  │                      │
                  │                      │ fail
                  │                      ▼
                  └─ reschedule ◄── attempts < max
                    (backoff)        attempts ≥ max ──► FAILED

Create — a JobInstance row is inserted (PENDING, attempts=0). This happens either from a template (via JobTemplateScheduler) or ad-hoc.
Schedule — jobExecutor.enqueue(job) emits job:created. The scheduler atomically claims the instance and enqueues it immediately if scheduledAt <= now, else arms a timer.
Process — p-queue calls JobWorker.processJob under concurrency.
Retry — failures before maxAttempts flip back to PENDING with an exponential backoff and a job:rescheduled event; the scheduler re-arms.
Terminal — COMPLETED/FAILED rows stay in the job_instance table (status + completedAt/result/error set). History is queried by status filter; no archive table exists.

Every lifecycle event is also broadcast to admin clients over SSE as a job.stats.updated event (see §7).

4. Producing Jobs

There are two paths: templates (recurring / manual-trigger) and ad-hoc instances (event-driven).

A. Recurring templates (from service code or REST API)

Create a Job template via JobTemplateService (services/job-template.service.ts):

import { jobTemplateService } from "#services/job-template.service";

const template = await jobTemplateService.createTemplate({
  name: "session-sweep",                 // unique
  type: "session-sweep",                 // handler key
  description: "Delete expired sessions",
  cronExpression: "0 * * * *",           // optional; omit for manual-trigger only
  enabled: true,
  priority: "NORMAL",
  maxAttempts: 3,
  timeoutMs: 60_000,
});

The service computes nextRunAt via croner and calls rearmTemplateScheduler(). To run a template once without affecting the schedule, use jobTemplateService.triggerTemplate(id).

Templates are mutable (updateTemplate); changes to cron/enabled recompute nextRunAt and take effect from the next scheduled run.

B. Ad-hoc instances (event-driven, from service code)

For event-driven work (e.g. delivering notifications), create a JobInstance directly via JobInstanceService (services/job-instance.service.ts) or the repository inside a transaction:

import { jobInstanceRepository } from "#repositories/job-instance.repository";
import { jobExecutor } from "#states";

const job = await prisma.$transaction(async (tx) => {
  // ...create child rows in the same tx...
  return jobInstanceRepository.create(
    { type: "send-notification", payload, description }, tx,
  );
});
jobExecutor.enqueue(job); // notify the scheduler AFTER the tx commits

Always enqueue() after the transaction commits, so the worker never sees a job whose payload rows aren't visible yet. The notification service (notification.service.ts) follows this pattern.

C. Externally (REST API)

POST /api/jobs — create a job template.
POST /api/jobs/:id/trigger — run a template once immediately.
POST /api/job-instances — create an ad-hoc instance (one-shot, no template).

5. Adding a New Job Handler

Create the handler in states/job-executor/handlers/<name>.handler.ts:

import type { JobHandler } from "#lib/queues/job.types";

export interface MyPayload { foo: string }

export const myHandler: JobHandler = async (job) => {
  const payload = (job.payload ?? {}) as Partial<MyPayload>;
  if (!payload.foo) throw new Error("payload missing: foo");
  // ...do work, return a JSON-serialisable result...
  return { ok: true };
};

Register it in states/job-executor/handlers/index.ts:
```
registry.register("my-job-type", myHandler);
```
(For recurring work) Seed a Job template with the matching type and a cronExpression in src/seed.ts (builtInJobTemplates), or create one via the admin API/JobTemplateService.

The handler is keyed by the string passed as the instance's type. Throwing rejects the instance (subject to retry/backoff); returning a value stores it as result.

6. Built-in Handlers

Six handlers ship today (handlers/index.ts):

send-notification — when notifications are created from a template (createNotificationsFromTemplate), a single send-notification instance carrying the notificationIds is created in the same transaction as the notification rows (jobId = null, ad-hoc). The handler calls deliverNotifications, dispatching per provider (in-app, smtp-email, …).
session-sweep — deletes rows where revokedAt IS NOT NULL or expiresAt < now(). Seeded as a recurring template (cron "0 * * * *", hourly) in src/seed.ts (builtInJobTemplates).
job-instance-sweep — deletes COMPLETED and FAILED instances older than 30 days. Seeded with cron "0 3 * * *" (daily at 3:00 AM).
verification-sweep — deletes expired verification tokens. Seeded with cron "30 * * * *" (hourly at :30).
operation-log-sweep — deletes operation logs older than 30 days. Seeded with cron "15 3 * * *" (daily at 3:15 AM).
audit-log-sweep — deletes audit logs older than 180 days. Seeded with cron "30 3 * * *" (daily at 3:30 AM).

All five recurring templates are seeded via builtInJobTemplates in packages/service/src/seed.ts.

7. Observability

SSE push: states/job-executor/index.ts subscribes to the executor and publishes job.stats.updated (target sse:admin:*:*) on every lifecycle event. Admin clients receive it via GET /api/events (routes/events/streamEvents.ts).
Stats endpoint: GET /api/jobs/stats returns live runtime numbers (queueSize, pending, concurrency) plus DB aggregates grouped by status and the next scheduled time.
Listing: GET /api/jobs (templates) and GET /api/job-instances (instances, filterable by status/type/jobId), both paginated.
Control: DELETE /api/jobs/:id (delete template), PATCH /api/jobs/:id (edit template), DELETE /api/job-instances/:id (cancel pending instance).

8. Configuration & Caveats

Environment variables

JOB_CONCURRENCY — max in-flight instances for the p-queue (default 5).
Template/instance-level overrides: maxAttempts (3), timeoutMs (60 000), priority (NORMAL).

Startup: jobExecutor.start() is invoked at module load in src/app.ts. On boot it runs recoverStuckProcessing() (resets PROCESSING instances orphaned by a previous/crashed process back to PENDING), then loadExpiredJobs() (atomically claims and enqueues due instances), and dispatchDue() (dispatches due templates).

Important limitations to keep in mind:

Single-process constraint narrowed to boot recovery. Instance dispatch is now claim-protected: claimDueJobs and claimJobById both flip PENDING → PROCESSING atomically (via SELECT … FOR UPDATE SKIP LOCKED and conditional updateMany respectively), so concurrent workers within a single process cannot duplicate-execute the same instance. However, recoverStuckProcessing() unconditionally resets all PROCESSING rows to PENDING at boot — safe only when one process owns the queue. A second replica or rolling deploy with overlap would reset the first's in-flight jobs, causing concurrent re-execution. The in-memory p-queue, timers, and lifecycle events also remain process-local. Do not horizontally scale without first replacing the boot reset with a stale-claim sweep keyed on a heartbeat/lease column (e.g. claimAt/startedAt + a periodic sweep that re-queues rows whose claim is older than a TTL).
Retention via sweep handlers. The job-instance-sweep handler deletes terminal instances (COMPLETED/FAILED) older than 30 days on a daily schedule. Other sweep handlers prune operation logs (30 days), verification tokens, and audit logs (180 days) — see §6 for the full list.
Priority is informational. priority is stored and surfaced in the API but does not influence execution order — p-queue runs in FIFO insertion order.
Timer ceiling. Delays beyond 24h are served by chained 24h timers; a process restart before the final fire re-covers them via loadExpiredJobs() / dispatchDue().

Architecture: Cache System

The platform has a lightweight in-memory LRU cache for hot, read-heavy records (currently notification channels and templates). There is no Redis — it is a process-local cache built on the lru-cache npm package, living inside the same Node process that serves the Hono API.

All cache code lives under packages/service/src/lib/cache.ts (the primitive), packages/service/src/states/cache.ts (the shared singletons), and packages/service/src/services/cache.service.ts (the admin inspection layer).

1. The `Cache` primitive — `lib/cache.ts`

A small wrapper around LRUCache<string, object> with one notable feature: namespacing.

static create(maxSize = 1000): Cache   // root cache, owns the LRU instance
namespace(ns: string): Cache           // returns a new view over the SAME LRU,
                                       //   with all keys prefixed "ns:"

A namespace is not a copy — it shares the underlying LRU instance with its parent, so the max budget is global. namespace() composes: calling it on an already-namespaced cache extends the prefix (a:b:key). Methods on a namespaced view automatically prepend the prefix, and clear()/keys() are scoped to the current namespace.

Method	Behaviour
`get<T>(key)`	Returns cached value or `undefined`
`set(key, value)`	Stores a value (no TTL — see §5)
`delete(key)`	Removes one key
`clear(prefix?)`	Clears keys beginning with `prefix`; with no arg, clears the whole current namespace (or the entire LRU if unnamespaced)
`getOrSet<T>(key, fn)`	Cache-aside helper: `get`, on miss call `fn`, `set`, return
`keys()` / `size`	Keys (namespaced) / count
`has(key)`	Existence check

2. The shared singletons — `states/cache.ts`

One root cache is created at startup and partitioned into named namespaces:

export const globalCache = Cache.create(CACHE_MAX_SIZE);            // env, default 1000
export const notificationChannelCache  = globalCache.namespace("notification:channel");
export const notificationTemplateCache = globalCache.namespace("notification:template");

Because they share globalCache's LRU instance, the three exports compete for a single max budget of CACHE_MAX_SIZE entries (default 1000).

3. How data flows (cache-aside)

The system uses a manual cache-aside pattern in the notification services. There is no automatic wiring; each consumer reads, misses, fetches, and stores:

Read (getActiveNotificationChannel, channel.service.ts:226):

const cached = notificationChannelCache.get<Channel>(id);
if (cached) return cached;            // hit

const channel = await prisma.notificationChannel.findFirst({ ... });
notificationChannelCache.set(id, channel);   // populate
return channel;

Invalidation (write-through, on every mutation): create/update/delete on a channel or template calls notificationChannelCache.clear() / notificationTemplateCache.clear(). Note these clear the entire namespace (coarse-grained) rather than a single key — blunt but safe.

One subtlety in findTemplateForDelivery (template.service.ts:255): disabled templates/channels are never cached (the entry is only .set() when there is no disabledReason), so a later enable is always seen fresh.

4. The Admin Inspection API — `routes/cache/`

A full management surface (guarded by the cache::manage permission) lets operators inspect and manipulate the live cache:

Endpoint	Action
`GET /api/cache/stats`	Total keys, `maxSize`, per-namespace key counts
`GET /api/cache/keys?search=`	List keys (with substring search), split into namespace/key + value type
`GET /api/cache/entry?key=`	Read a single entry's value
`PUT /api/cache/entry`	Set/update an arbitrary entry (full key + value)
`DELETE /api/cache/entry?key=`	Delete one entry
`DELETE /api/cache/namespace`	Clear every key under a namespace
`DELETE /api/cache/all`	Wipe the whole cache (audited as `cache.all_cleared`)

The CacheService (services/cache.service.ts) derives namespace/key by splitting on the last : in the full key, so multi-segment namespaces like notification:channel:abc123 are reported as namespace notification:channel, key abc123. The admin UI (apps/admin/.../cache/components/cache-tree.tsx) renders these as a collapsible tree grouped by :-separated segments, with refresh, search, and clear actions.

5. Configuration & Caveats

Environment variables

CACHE_MAX_SIZE — global entry cap shared by all namespaces (default 1000).

Important limitations to keep in mind:

In-memory / single-process. The cache is not persisted and is not shared across processes. A restart empties it, and multiple API instances each hold an independent cache (no cross-instance coherence). This matches the Job queue's single-process assumption (see Job System §8).
No TTL. The LRU is configured with max only — there is no time-based expiry. Entries are evicted purely by count (once max is exceeded) or by explicit invalidation. Stale data therefore lives until evicted or cleared.
Coarse invalidation. Mutations .clear() the whole namespace rather than the affected key(s). This is safe but over-invalidates (every channel update flushes all channels).
getOrSet is unused. The cache-aside helper on Cache exists but has no callers today; consumers do manual get/set. It is also not stampede-safe — concurrent misses each fetch independently (no in-flight promise de-dup).
Permissive set/get typing. set(key, unknown) stores unknown, and get<T>() is an unchecked cast. A wrong T at the read site will compile but return garbage, so consumers must keep their read/write types aligned.

Architecture: Rate Limit System

The platform rate-limits HTTP requests with a fixed-window, in-process limiter. There is no Redis — counters live in memory inside the Hono API process. The system layers a global limiter over a stricter auth limiter, and adds per-subject overrides (whitelisting, custom ceilings) that admins can tune at runtime through the API.

All rate-limit code lives under packages/service/src/lib/rate-limit-store.ts (counter storage), lib/rate-limit-registry.ts (limiter + override registry), middleware/rate-limit.ts (Hono middleware), services/rate-limit.service.ts (admin/config layer), and routes/rate-limit/ (management endpoints).

1. Core Components

  request ──► createRateLimiter (middleware)
                  │  1. resolve subject (user:<id> | ip:<addr>)
                  │  2. rateLimitRegistry.resolvePolicy(name, subject)
                  │  3. store.hit(subject, windowMs)
                  ▼
            RateLimitStore          ◄──── RateLimitRegistry (name → limiter)
            (Map subject→bucket)            • base max / windowMs
            • hit() · reset()              • SystemConfig defaults (runtime)
            • sweeper interval              • RateLimitOverride rows (per subject)

RateLimitStore — `lib/rate-limit-store.ts`

A Map<string, { count, resetAt }> keyed by subject string. hit(key, windowMs) is a first-hit-anchored fixed window: the first request in a lull sets resetAt = now + windowMs; subsequent hits increment count; once resetAt passes, the next hit starts a fresh window. A background sweeper (setInterval, 60 s, .unref()'d) evicts expired buckets so the map doesn't grow unbounded. Each limiter owns its own store instance.

RateLimitRegistry — `lib/rate-limit-registry.ts`

A singleton holding the named limiters and two override sources. Its key method is resolvePolicy(name, subject), which composes a final policy in this order:

Base — the limiter's registered max / windowMs / enabled.
SystemConfig defaults — applied live via updateDefaults(name, {...}) (called at boot from the rate-limit config group, see §3).
Per-subject override — a RateLimitOverride DB row for this exact subject (if present and within its startAt/endAt window). An override can supply a custom max/windowMs or set bypass: true to whitelist the subject.

It also exposes snapshot() for status inspection and releaseKey() / releaseSubject() to manually clear a subject's bucket.

createRateLimiter — `middleware/rate-limit.ts`

Hono middleware factory (createRateLimiter({ name, max, windowMs, enabled })). On each request it:

Resolves the subject: user:<id> when a session cookie is present, otherwise ip:<first x-forwarded-for | x-real-ip | "unknown">.
Looks up the limiter entry by name; if disabled, calls next().
resolvePolicy → store.hit → sets X-RateLimit-Limit, -Remaining, -Reset headers.
If count > max: returns 429 { code: 429, message: "Too Many Requests" } with a Retry-After header, and (on the first over-limit hit) broadcasts a rate_limit.updated SSE event.

rate-limit.service.ts

The admin/config layer. Loads overrides and defaults at boot, and exposes: status snapshots, override CRUD, and a manual "release" (counter reset).

2. How Limiters Are Wired — `src/app.ts`

Two limiters are created and mounted in app.ts:79-97:

Limiter	Mounted on	Default limit	Env defaults
`global`	`*` (every route)	300 req / 60 s	`RATE_LIMIT_GLOBAL_MAX`, `RATE_LIMIT_GLOBAL_WINDOW_MS`
`auth`	`/auth/sign-in/email`, `/auth/sign-up/email`, `/auth/sign-in/wechat`, `/auth/change-password`	10 req / 60 s	`RATE_LIMIT_AUTH_MAX`, `RATE_LIMIT_AUTH_WINDOW_MS`

Both are gated by the global RATE_LIMIT_ENABLED flag (default on). Because the auth routes are also under *, a single auth request consumes a bucket in both the auth store and the global store (they are independent stores).

3. Configuration Sources (3 layers)

Policy resolution merges three sources, checked innermost-wins:

Env vars — seed the limiter at construction (the table above).
SystemConfig group rate-limit — keys enabled, global.max, global.windowMs, auth.max, auth.windowMs. Loaded at boot by initRateLimitDefaults() (app.ts:29) and applied live via rateLimitRegistry.updateDefaults(). Editing these configs + calling reloadRateLimitDefaults() changes limits without a restart, and the store references are preserved (existing buckets survive).
RateLimitOverride table — per-subject rows (ip:x.x.x.x or user:<id>). Loaded into memory at boot by initRateLimitOverrides() (app.ts:32). Each row can set a custom max/windowMs, bypass (whitelist), a note, and an optional startAt/endAt validity window (e.g. a temporary lift during a demo). Rows mutated through the API are applied live via setOverride/removeOverride.

4. Admin Management API — `routes/rate-limit/`

All endpoints require the rate-limit::manage permission:

Endpoint	Action
`GET /api/rate-limit/status?limiter=&blocked=`	Live bucket snapshot: subject, count, remaining, blocked, resetAt. Filterable to one limiter and/or blocked-only.
`GET /api/rate-limit/settings`	Configured limiters (name/max/windowMs).
`GET /api/rate-limit/overrides`	List all per-subject overrides.
`PUT /api/rate-limit/overrides/:subject`	Upsert an override (persisted to DB and applied live); broadcasts `rate_limit.updated`.
`DELETE /api/rate-limit/overrides/:subject`	Remove an override (persisted + live).
`POST /api/rate-limit/release`	Manually reset a subject's counter — scope to one `limiter` or all. Audited as `rate_limit.released`.

5. Configuration & Caveats

Environment variables

RATE_LIMIT_ENABLED — global kill switch (default on; set false to disable).
RATE_LIMIT_GLOBAL_MAX / RATE_LIMIT_GLOBAL_WINDOW_MS — global limiter (300 / 60 000 ms).
RATE_LIMIT_AUTH_MAX / RATE_LIMIT_AUTH_WINDOW_MS — auth limiter (10 / 60 000 ms).

Important limitations to keep in mind:

In-memory / single-process. Counters are not shared across instances. Behind a load balancer with N API instances, the effective per-subject limit is roughly N × max, since each instance counts independently (a subject's requests are hashed/routed across instances). This mirrors the Job queue and Cache single-process assumptions.
Fixed-window boundary bursts. The window is anchored to the first request, not to clock alignment. A burst of max requests just before resetAt, followed by another max immediately after, yields ~2 × max requests inside one windowMs span. This is acceptable for abuse prevention but not a hard guarantee.
X-Forwarded-For is trusted blindly. The subject IP is taken as the first x-forwarded-for entry with no validation. If the API is reachable without a proxy that overwrites/normalises that header, a client can spoof its IP to evade limits. Ensure all ingress goes through a trusted proxy (or validate the header chain) before relying on IP-based limits.
SSE routing is target-based. EventBus (lib/event-bus.ts) is a generic topic router: subscribers register targets (:-joined, single-segment * wildcard, symmetric), and publish/close match by event.target. Each SSE connection subscribes under sse:<appCode>:<userId>:<token>; publishers fan out with wildcards — sse:admin:*:* (admin dashboard, e.g. rate_limit.updated / job.stats.updated), sse:<appCode>:<userId>:* (an in-app notification), or sse:*:<userId>:* (an app-agnostic notification). signOut resolves the app via requireCurrentApp and calls close("sse:<appCode>:<userId>:<token>").
Direct DB override writes need a restart. Overrides loaded at boot are kept in sync only when mutated through the API (which calls setOverride live). A row added directly to RateLimitOverride won't take effect until restart.

Architecture: SSE Event Bus

The platform includes a generic, topic-based pub/sub event bus for real-time Server-Sent Events (SSE) push to connected clients. It is used by the job executor, rate limiter, and sign-out flows to broadcast state changes to admin dashboards and in-app notification inboxes.

Event bus code lives in packages/service/src/lib/event-bus.ts. SSE connections are served by routes/events/streamEvents.ts at GET /api/events.

1. Topic Routing

The event bus routes messages by target strings — colon-delimited paths with single-segment * wildcards. Both publishers and subscribers specify targets; matching is symmetric.

Pattern	Matches
`sse:admin::`	Any admin connection (all users, all tokens)
`sse:<appCode>:<userId>:*`	All of a user's connections within one app
`sse::<userId>:`	All of a user's connections across all apps
`sse:<appCode>:<userId>:<token>`	One specific connection

Publishers use wildcards to fan out:

sse:admin:*:* — admin dashboard events (job.stats.updated, rate_limit.updated)
sse:<appCode>:<userId>:* — in-app notifications for a specific user
sse:*:<userId>:* — app-agnostic notifications (e.g. sign-out)

Subscribers (SSE connections) register under their exact identity: sse:<appCode>:<userId>:<token>.

2. Internal Design

The bus indexes subscribers by app-code segment for O(1+app_count) fan-out instead of O(subscribers). publish(target, data, event?) walks the index, and close(target) removes subscribers (used by signOut to disconnect a specific user's SSE connection).

3. Consumers

Producer	Event	Target
Job Executor (§7)	`job.stats.updated`	`sse:admin::`
Rate Limiter (§4)	`rate_limit.updated`	`sse:admin::`
Sign-out	(close)	`sse:<appCode>:<userId>:<token>`

Admin clients receive these events via the useEventStream hook from @repo/frontend, which manages the EventSource connection per app, and the organization portal uses it for in-app notification delivery.

Architecture: Audit Log System

The platform records structured audit entries for security-relevant mutations across the API. Every audited action captures the actor, target resource, before/after state, and outcome.

Audit log code: routes/audit-log/, services/audit-log.service.ts, prisma/schema.prisma (AuditLog model).

1. Data Model

The AuditLog table stores:

Field	Purpose
`id`	cuid
`traceId`	Correlates related operations (from trace context middleware)
`userId`	Actor who performed the action
`event`	Action identifier (e.g. `user.created`, `role.assigned`)
`category`	Grouping (e.g. `user`, `role`, `organization`)
`severity`	`info` / `warning` / `error`
`outcome`	`success` / `failure`
`resourceType` / `resourceId`	Target entity
`before` / `after`	JSON diff of state change
`metadata`	Additional context (IP, user agent, etc.)
`createdAt`	Timestamp

2. How Audits Are Written

Audits are written via AuditLogService.createEntry() called from route handlers or service methods. The method accepts structured fields and serialises the before/after diffs as JSON.

3. Retention

The audit-log-sweep job handler deletes entries older than 180 days (seeded as a recurring template with cron "30 3 * * *", daily at 3:30 AM).

4. API & UI

GET /api/audit-logs — paginated listing, filterable by event/category/severity/userId/date range.
Admin UI page at /audit-logs with filters and detail view.

Architecture: Operation Log System

Operation logs provide an automatic, per-request trace of every HTTP call to the API — complementing the explicit, action-scoped audit logs.

Operation log code: routes/operation-log/, services/operation-log.service.ts, middleware/operation-logger.ts, middleware/trace-context.ts.

1. Automatic Logging Middleware

The operationLogger middleware wraps every API route. For each request it records:

Field	Source
`traceId`	Per-request UUID from `traceContext` middleware
`method` / `path`	HTTP method and route path
`statusCode`	Response status
`durationMs`	Wall-clock request time
`ip`	Client IP (from `x-forwarded-for` / `x-real-ip`)
`authType` / `authTokenId`	Session or API token used, if any
`errorName` / `errorMessage` / `stack`	When the response status is ≥ 400
`level`	`info` / `warn` / `error` based on status code
`source` / `module`	Calling app code and route module

The middleware uses Hono's response streaming hooks to capture the final status code and duration even for streaming/SSE responses.

2. Retention

The operation-log-sweep job handler deletes entries older than 30 days (seeded with cron "15 3 * * *", daily at 3:15 AM).

3. API & UI

GET /api/operation-logs — paginated, filterable by level/source/module/method/path/statusCode/date range.
Admin UI page at /operation-logs with filters, detail view, and trace ID linking to related audit logs.

Architecture: System Config System

System configuration is stored in the database with JSON Schema validation, surfaced through an admin UI, and layered over environment variables via a mechanical fallback mechanism.

Config code: routes/system-config/, services/system-config.service.ts, services/system-config-env.service.ts, prisma/schema.prisma (SystemConfig model).

1. Data Model

The SystemConfig table stores key-value configs grouped by category:

Field	Purpose
`id`	cuid
`group`	Config category (e.g. `auth`, `rate-limit`, `upload`, `webauthn`)
`key`	Config key within the group
`value`	JSON-encoded value
`type`	JSON Schema type constraint
`schema`	JSON Schema for validation and admin UI rendering
`label` / `description`	Human-readable metadata
`isSecret`	Masks value display in the admin UI
`sortOrder`	Ordering hint for UI tabs

2. Env Fallback Mechanism

system-config-env.service.ts provides mergeEnvFallback(group, key) which mechanically derives an env var name from group + key (e.g. auth + enabled → AUTH_ENABLED). If the DB row has no value, the env var is used as a fallback. This powers runtime-configurable subsystems like rate limiting and WebAuthn.

3. Config Groups

Group	Purpose
`auth`	Registration enabled, password policies
`webauthn`	RP ID, origin overrides
`upload`	Sign secret, file size limits, hotlink domains
`rate-limit`	Global and auth limiter defaults (see Rate Limit §3)

4. Admin UI

System configs are managed through the admin UI under the Settings page as a tabbed form. Each group renders dynamic inputs driven by the JSON Schema, with secret masking for sensitive values. Changes are applied live without a restart for runtime subsystems that subscribe to config updates.

Architecture: File Upload & Attachment System

The platform provides a full file upload pipeline with hash-based deduplication, signed URL access control, hotlink protection, and polymorphic attachment associations.

Upload/attachment code: routes/attachment/, services/attachment.service.ts, middleware/body-limit.ts, prisma/schema.prisma (Upload, Attachment models).

1. Data Model

Two models form the pipeline:

Upload — the physical file record:

Field	Purpose
`id`	cuid
`path`	Sharded storage path (`hash[0:2]/hash[2:4]/hash.ext`)
`mimeType`	Validated MIME type
`size`	File size in bytes
`hash`	SHA-256 hash for deduplication
`createdAt`	Timestamp

Attachment — the logical reference:

Field	Purpose
`id`	cuid
`bizType` / `bizId`	Polymorphic business entity reference
`uploadId`	FK to `Upload`
`visibility`	`public` / `private`
`createdBy`	FK to `User`

2. Upload Pipeline

Create attachment — POST /api/attachments returns a signed upload URL and an uploadId/attachmentId pair.
Upload — the client PUTs the file to the signed URL. The server:
- Validates MIME type (magic byte check, not just Content-Type header).
- Computes SHA-256 hash.
- If the hash already exists in Upload, reuses the existing file record (deduplication) — the new Attachment points to the existing Upload.
- Otherwise, writes the file to the sharded storage path and creates a new Upload record.
Serve — files are served through GET /api/attachments/:id with optional ?token=&expires= for signed private access.

3. Signed URLs & Hotlink Protection

Private attachments require a signed token (?token= + ?expires=) generated via HMAC-SHA256 with UPLOAD_SIGN_SECRET. Public attachments can be hotlink-protected by domain:

UPLOAD_HOTLINK_ENABLED — toggle (default false).
UPLOAD_HOTLINK_ALLOWED_DOMAINS — comma-separated domains.
UPLOAD_HOTLINK_ALLOW_EMPTY_REFERER — allow direct access (browser URL bar).

Requests with a disallowed Referer header receive a 403.

4. Size Limits

body-limit.ts middleware distinguishes multipart uploads from JSON requests:

MAX_UPLOAD_FILE_SIZE — individual file cap (default 5 MB).
MAX_UPLOAD_BODY_SIZE — total multipart body cap.

5. Admin Management

GET /api/attachments — paginated listing, filterable by visibility, MIME type, uploader, date range.
PUT /api/attachments/:id/replace — replace an attachment's file in-place (creates a new Upload, updates the Attachment.uploadId).
DELETE /api/attachments — batch delete by IDs.
Admin UI at /attachments with file previews and filters.

Architecture: AI Agent System

The platform includes an in-app AI chat assistant with tool-calling capabilities. It can execute platform API operations and read files, gated by per-application configuration and admin-controlled API allowlists.

Agent code: routes/agent/, services/agent-config.service.ts, services/agent-session.service.ts, lib/ai-agent/, prisma/schema.prisma (AgentSession, AgentMessage models). Frontend: packages/frontend/src/components/agent-launcher/, packages/frontend/src/components/agent-chat/, packages/frontend/src/hooks/use-agent-chat.ts.

1. Data Model

AgentSession — a chat conversation:

Field	Purpose
`id`	cuid
`userId`	FK to `User` — session owner
`appId`	FK to `Application` — scoping
`name`	Auto-generated title from first message (updated via SSE)
`createdAt`	Timestamp

AgentMessage — a message within a session:

Field	Purpose
`id`	cuid
`sessionId`	FK to `AgentSession`
`role`	`user` / `assistant`
`parts`	JSON — AI SDK `UIMessagePart[]`
`createdAt`	Timestamp

2. API Endpoints

Endpoint	Purpose
`GET /api/agent/sessions`	List sessions (paginated)
`POST /api/agent/sessions`	Create a new session
`GET /api/agent/sessions/:id`	Get session with full message history
`DELETE /api/agent/sessions/:id`	Delete a session
`POST /api/agent/sessions/:id/messages`	Send a message (streaming response via AI SDK)
`POST /api/agent/sessions/:id/files`	Upload a file for context

3. Tool System

The agent is equipped with tools registered in lib/ai-agent/tools/:

Tool	Purpose
`call_api`	Execute arbitrary platform API endpoints (read operations). Gated by the application's allowed API list (see §4).
`get_api_schema`	Retrieve the OpenAPI spec to discover available operations.
`read_file`	Read uploaded files associated with the session.

Tools are defined as AI SDK tool objects and run server-side within the streaming handler.

4. Configuration & Access Control

AI provider config is per-application via ApplicationConfig (group ai-agent): baseURL, apiKey, model, reasoning. Falls back to env vars: AI_AGENT_BASE_URL, AI_AGENT_API_KEY, AI_AGENT_MODEL.

Permission gating: the agent launcher in the admin sidebar is shown only when the user holds system/agent:chat.

Allowed API selector: admins configure which API operations the agent's call_api tool may invoke via POST /api/applications/:id/allowed-apis. The openapi.service.ts parses the OpenAPI spec to list available operations; selected operations are stored per application. At runtime, call_api validates the target operation against this allowlist.

5. Frontend Integration

The agent launcher lives in the admin sidebar as a floating chat button. It opens a resizable panel with:

Session list (create, select, delete, lazy-load, infinite scroll)
Streaming chat with markdown rendering
File upload with drag-and-drop
Tool call cards (rendered inline in the message stream)

The chat uses the AI SDK's useChat hook with DefaultChatTransport, streaming responses from POST /api/agent/sessions/:id/messages. Session titles are updated live via SSE (agent.session.title.updated).

The AgentPanel component (@repo/frontend) is display-mode agnostic and can be embedded in a Sheet, a full page, or any flex container.

Architecture: Permission & RBAC System

The platform implements a scoped, hierarchical Role-Based Access Control (RBAC) system with group::action permission codes, menu-to-permission gating, position-based role assignment, and API token scoping.

Permission code: packages/shared/src/permissions.ts. Service code: routes/permission/, routes/role/, routes/menu/, services/permission.service.ts, services/role.service.ts.

1. Permission Codes

Permissions follow the group::action format:

Example	Meaning
`user::read`	View user lists and details
`user::write`	Create/update/delete users
`role::assign`	Assign roles to users
`system/agent:chat`	Access the AI agent chat
`cache::manage`	Manage the runtime cache
`rate-limit::manage`	Manage rate limit overrides

Permissions are defined in packages/shared/src/permissions.ts and seeded into the database. Each permission belongs to an application scope.

2. Role Assignment

Roles are collections of permissions, scoped per application:

Role — has a name, appId, and a set of RolePermission rows.
RoleAssignment — links a Role to a User within an organization context.
A user's effective permissions are the union of all roles assigned to them across all organizations and apps.

Additionally, roles can be assigned to Position records (Position.roleId). Members holding a position inherit the position's role permissions — enabling department-level permission management without per-user role assignment.

3. Menu Permission Gating

Admin menus (Menu model) are associated with permissions via MenuPermission. A menu item is only visible to a user if they hold at least one of the menu's associated permissions. This drives the sidebar navigation in the admin UI.

4. API Token Scopes

API tokens (ApiToken) can be restricted to a subset of permissions via the scopes field. When a request is authenticated with an API token, the token's scopes are intersected with the user's effective permissions.

5. Enforcement

Route handlers call requirePermission(permission) / requirePermissions([]) (from middleware/permission.ts), which resolves the current user's effective permissions (accounting for all role assignments + token scopes) and throws 403 if the required permission is absent.

Architecture: Notification System

The platform dispatches notifications through configurable channels with template-based message rendering, delivery tracking, and background job integration.

Notification code: routes/notification/, services/notification.service.ts, services/notification-channel.service.ts, services/notification-template.service.ts, jobs via send-notification handler.

1. Data Model

NotificationChannel — a delivery method:

Field	Purpose
`id`	cuid
`type`	`in-app` / `smtp-email` / `sms`
`name`	Human-readable label
`config`	JSON — channel-specific settings (SMTP server, SMS provider)
`enabled`	Toggle
`appId`	FK to `Application`

NotificationTemplate — a reusable message template:

Field	Purpose
`id`	cuid
`name` / `description`	Human-readable metadata
`subject`	Template subject with `{{variable}}` placeholders
`body`	Template body with `{{variable}}` placeholders
`channels`	Array of channel IDs to deliver through

Notification — a concrete message instance:

Field	Purpose
`id`	cuid
`userId`	Recipient
`templateId`	Source template (nullable for ad-hoc)
`subject` / `body`	Rendered message
`status`	`pending` / `sent` / `failed`
`channel`	Delivery channel used
`deliveredAt`	When successfully sent

2. Dispatch Flow

Create — createNotificationsFromTemplate inserts Notification rows (rendering template variables) and creates a JobInstance (type send-notification, notificationIds) in the same transaction.
Enqueue — jobExecutor.enqueue() is called after the transaction commits, ensuring the worker never sees notification rows before they're visible.
Deliver — the send-notification job handler calls deliverNotifications, which iterates over each notification and dispatches via the matching channel provider (in-app, smtp-email, …).
Track — each Notification row is updated with status and deliveredAt.

3. Channel Providers

Channel	Implementation
`in-app`	Stored in DB, surfaced in the notification inbox UI
`smtp-email`	Nodemailer with configurable SMTP settings
`sms`	Outbox pattern — stored for external SMS provider pickup

4. Caching

Notification channels and templates are cached in the LRU cache (see Cache System). Cache-aside reads in channel.service.ts and template.service.ts avoid repeated DB lookups during dispatch. Disabled templates/channels are never cached, so enabling them is always seen fresh. Mutations clear the entire namespace (coarse-grained invalidation).

5. Admin UI

Channel management with test-send workflow (POST /api/notification/channels/:id/test).
Template editor with variable rendering preview.
Notification record history with status/date filtering.

Architecture: API Token System

Users can create personal API tokens for programmatic access to the API, with scoped permissions, expiration, and usage tracking.

Token code: routes/token/, services/api-token.service.ts, prisma/schema.prisma (ApiToken model).

1. Data Model

Field	Purpose
`id`	cuid
`userId`	FK to `User` — token owner
`name`	Human-readable label
`tokenHash`	SHA-256 of the full token (the full token is shown only once at creation)
`tokenPrefix` / `tokenSuffix`	First/last 8 chars for identification in listings
`scopes`	JSON array of `group::action` permission codes
`expiresAt`	Optional expiry
`lastUsedAt`	Auto-updated on each authenticated request
`createdAt`	Timestamp

2. Authentication Flow

Requests can authenticate via Authorization: Bearer <token>. The bearer auth middleware:

Extracts the token string.
Hashes it with SHA-256.
Looks up ApiToken by tokenHash.
Validates expiry.
Resolves the user and scopes — effective permissions are the intersection of the user's role permissions and the token's scopes.

3. Token Management

POST /api/tokens — create a token (returns the full token once).
GET /api/tokens — list tokens (prefix/suffix only, never the full token).
DELETE /api/tokens/:id — revoke a token.
Admin and organization apps both have token management UIs.

FilesExpand file tree

ARCHITECTURE.md

Latest commit

History

ARCHITECTURE.md

File metadata and controls

Architecture

Architecture: Job System

1. Data Model

2. Core Components

JobQueue — lib/queues/job-queue.ts

JobScheduler — lib/queues/job-scheduler.ts (instance scheduler)

JobTemplateScheduler — lib/queues/job-template-scheduler.ts (recurring engine)

JobWorker — lib/queues/job-worker.ts

JobHandlerRegistry — lib/queues/job-handler-registry.ts

JobExecutorContext — lib/queues/job-executor-context.ts

JobExecutor — lib/queues/job-executor.ts

3. Lifecycle of a Job Instance

4. Producing Jobs

A. Recurring templates (from service code or REST API)

B. Ad-hoc instances (event-driven, from service code)

C. Externally (REST API)

5. Adding a New Job Handler

6. Built-in Handlers

7. Observability

8. Configuration & Caveats

Architecture: Cache System

1. The Cache primitive — lib/cache.ts

2. The shared singletons — states/cache.ts

3. How data flows (cache-aside)

4. The Admin Inspection API — routes/cache/

5. Configuration & Caveats

Architecture: Rate Limit System

1. Core Components

RateLimitStore — lib/rate-limit-store.ts

RateLimitRegistry — lib/rate-limit-registry.ts

createRateLimiter — middleware/rate-limit.ts

rate-limit.service.ts

2. How Limiters Are Wired — src/app.ts

3. Configuration Sources (3 layers)

4. Admin Management API — routes/rate-limit/

5. Configuration & Caveats

Architecture: SSE Event Bus

1. Topic Routing

2. Internal Design

3. Consumers

Architecture: Audit Log System

1. Data Model

2. How Audits Are Written

3. Retention

4. API & UI

Architecture: Operation Log System

1. Automatic Logging Middleware

2. Retention

3. API & UI

Architecture: System Config System

1. Data Model

2. Env Fallback Mechanism

3. Config Groups

4. Admin UI

Architecture: File Upload & Attachment System

1. Data Model

2. Upload Pipeline

3. Signed URLs & Hotlink Protection

4. Size Limits

5. Admin Management

Architecture: AI Agent System

1. Data Model

2. API Endpoints

3. Tool System

4. Configuration & Access Control

5. Frontend Integration

Architecture: Permission & RBAC System

1. Permission Codes

2. Role Assignment

3. Menu Permission Gating

4. API Token Scopes

5. Enforcement

Architecture: Notification System

1. Data Model

JobQueue — `lib/queues/job-queue.ts`

JobScheduler — `lib/queues/job-scheduler.ts` (instance scheduler)

JobTemplateScheduler — `lib/queues/job-template-scheduler.ts` (recurring engine)

JobWorker — `lib/queues/job-worker.ts`

JobHandlerRegistry — `lib/queues/job-handler-registry.ts`

JobExecutorContext — `lib/queues/job-executor-context.ts`

JobExecutor — `lib/queues/job-executor.ts`

1. The `Cache` primitive — `lib/cache.ts`

2. The shared singletons — `states/cache.ts`

4. The Admin Inspection API — `routes/cache/`

RateLimitStore — `lib/rate-limit-store.ts`

RateLimitRegistry — `lib/rate-limit-registry.ts`

createRateLimiter — `middleware/rate-limit.ts`

2. How Limiters Are Wired — `src/app.ts`

4. Admin Management API — `routes/rate-limit/`