03 / Develop · Build · Cross-system

Cross-system

Layers 5–6 of six (the map) — the calendar works; now let it meet the rest of the platform.

Where the calendar meets the rest of the platform: reaching data that lives in another system, and keeping one organization's data invisible to another. Both are boundaries — one between apps, one between tenants.

bash

veni enable integration    # records the capability — writes no files
veni enable tenancy        # records the capability — writes no files

Both are developer-authored, like authz: they mark the capability in veni.json and print the steps. Everything below is code you write, against pieces the earlier layers already gave you.

Integration — invite people

Requirement

Inviting a colleague means picking them from a list of real people. The Shell owns that list — and the picker must open instantly, even on a morning when the Shell is down.

Resolving it. A satellite never reads another system's database, so you get users one of two ways:

Keep a copy (events) — subscribe to the Shell's shell:stream:users (Redis Streams) and maintain your own people table, so the invite picker is instant and still works when the Shell is down. This is the default, and for the Shell's directory it is effectively the only standing one.
Ask now (HTTP) — for one fresh answer, call the owning system directly. Between satellites that's Connect RPC, which the scaffold is set up for both ways. Against the Shell it's its REST API — step 6 below has the one endpoint you need and what it costs.

The Shell exposes no RPC for reading users

Its Connect handler serves exactly one path, auth.v1.AuthGateway/ExchangeForService, and returns everything else to the router (grpc/connect-handler.ts). There is no ListUsers to call. Reach the directory through the stream, and through GET /api/users for the initial fill.

The scaffold ships no consumer — you add it. The shape: a local people table keyed by the Shell's user id, and a consumer that reads shell:stream:users under a group, upserts on user.created / updated / provisioned and removes on user.deleted, acking after each write. Two things bite: a group created at '$' starts empty (backfill once via RPC), and delivery is at-least-once (handlers must be idempotent).

Build the consumer — the full pattern

1 · Point at Redis. Already in api/.env (the token blacklist uses it) — the bus is the same Redis, on the host-mapped port (:16379, not the container's :6379):

bash

APP_ENV_REDIS_URL=redis://localhost:16379

RedisHelper takes parts, not a URL, so split it once — the same thing the Shell's publisher does:

import { RedisHelper } from '@venizia/ignis-helpers'   // NOT '@venizia/ignis'
const url = new URL(process.env.APP_ENV_REDIS_URL!)
const redis = new RedisHelper({
  name: 'calendar-api-consumer',
  host: url.hostname,
  port: url.port ? parseInt(url.port) : 6379,
  password: url.password || '',
  onError: ({ error }) => this.logger.warn('redis error (non-fatal)', error),
})
const client = redis.getClient()

2 · Know the contract. The Shell publishes with XADD as four flat fields. Depend on the contract, not the Shell's code — copy the stream names into your own src/events/registry.ts:

Field	Value
`event`	`user.created` · `user.updated` · `user.deleted` · `user.provisioned`
`payload`	the user, JSON-encoded as a string
`source`	`shell-api`
`timestamp`	ISO-8601

3 · A read-model table — your copy, keyed by the source id so a replayed event overwrites rather than duplicates:

export const people = pgTable('people', {
  id:        uuid('id').primaryKey(),        // the Shell's user id — NOT defaultRandom
  email:     varchar('email', { length: 320 }).notNull(),
  name:      varchar('name', { length: 255 }),
  updatedAt: timestamp('updated_at').notNull().defaultNow(),
})

4 · A consumer group — a private cursor + pending list, so replicas split the work. Poll '>' (never-delivered); a message you took but never acked sits in the pending list, so sweep it with XAUTOCLAIM on startup and a timer:

const GROUP = 'calendar-api'                    // one per app
const CONSUMER = `calendar-api-${process.pid}`  // one per replica

try {
  await client.xgroup('CREATE', 'shell:stream:users', GROUP, '$', 'MKSTREAM')
} catch (err: any) {
  if (!err.message?.includes('BUSYGROUP')) throw err   // already created
}

while (running) {
  const res = await client.xreadgroup(
    'GROUP', GROUP, CONSUMER, 'COUNT', 10, 'BLOCK', 5000, 'STREAMS', stream, '>',
  )
  for (const [s, messages] of res ?? []) for (const [id, f] of messages) await this.handle(s, id, f)
}

// Reclaim what a dead replica never finished. XAUTOCLAIM steals from any consumer
// idle > the limit — CONSUMER carries the pid, so a restarted pod can't see its own
// old pending list otherwise. The 60s floor avoids stealing live work.
const reply = await client.xautoclaim(stream, GROUP, CONSUMER, 60_000, '0', 'COUNT', 100)
for (const [id, f] of reply[1] ?? []) await this.handle(stream, id, f)

5 · Handle, then ack. Fields arrive as a flat [k, v, k, v]. Handlers must be idempotent (upsert, never insert) — delivery is at-least-once. Ack only after the write:

private async handle(stream: string, id: string, fields: string[]) {
  const data: Record<string, string> = {}
  for (let i = 0; i < fields.length; i += 2) data[fields[i]] = fields[i + 1]
  const payload = JSON.parse(data.payload ?? '{}')
  try {
    if (data.event === 'user.deleted') await this.people.remove(payload.id)
    else                               await this.people.upsert(payload)   // created + updated + provisioned
    await client.xack(stream, GROUP, id)     // success
  } catch (err) {
    this.logger.error(`failed ${data.event} (id=${id})`, err)  // no ack → the sweep retries it
  }
}

Start the loop from application.ts, stop it on shutdown.

6 · Backfill. A group created at '$' starts empty — everyone who signed up before your consumer's first boot is absent, and no event will ever arrive for them. Seed the table once from the Shell's REST directory, then let the stream keep it fresh:

GET {SHELL_API}/api/users        # e.g. http://localhost:3000/api/users
Authorization: Bearer <a Shell token>

Three constraints, and they decide when you can run it rather than how:

It needs the shell:user:list permission, seeded only to system-admin, admin and org-admin. An ordinary end-user's token gets a 403, so you cannot backfill lazily on any signed-in request.
It is org-scoped — results come back filtered by the requesting user's organization. One call fills one tenant.
There is no machine credential for it. Your API holds its own HS256 service token, which the Shell does not accept, and it doesn't retain the caller's Shell token after /api/auth/exchange.

So make it an admin-triggered one-off — a bun run backfill:people script, or a "Sync people" action behind requireAdmin(c) that forwards the calling admin's Shell token — and keep it re-runnable, since the upsert is idempotent. Do not put it in your boot path.

Best-effort bus. The Shell's publisher logs and moves on if Redis is down; it never throws, and a dropped event is never redelivered. Treat the stream as a fast path over a table you can always rebuild, and keep the backfill re-runnable.

Done when — the invite picker lists real people, and still works with the Shell stopped — it's reading your own local copy.

Multi-tenancy — org calendars

Requirement

Acme and Globex both use Calendar. Neither may ever glimpse the other's events — not in a list, not in a search, not by guessing an id.

Resolving it. Every token carries an organizationId — the tenant. Add the column to events, indexed, because from now on every query filters by it:

// src/models/schema.ts — add `index` to the drizzle-orm/pg-core import
export const events = pgTable('events', {
  // …id, title, startsAt, ...auditable
  organizationId: uuid('organization_id').notNull(),   // the tenant
}, (table) => ({
  orgIdIdx: index('events_organization_id_idx').on(table.organizationId),
}))

Re-run cd api && bun run db:generate && bun run db:migrate, then scope every query by the org:

const user = requireAuth(c)
if (!user.organizationId) throw …            // no tenant → no tenant-scoped write
const data = await this.events.listForOrg(user.organizationId, { from, to })

The auditable mixin already records who; the org column records whose tenant. Tenancy and full row-level isolation →

Adding NOT NULL to a table that already has rows fails

You created real events in the last layer, so db:generate emits ALTER TABLE "events" ADD COLUMN "organization_id" uuid NOT NULL and Postgres rejects it: column "organization_id" of relation "events" contains null values. Do it in three steps in the generated migration — add the column nullable, backfill the existing rows with the org they belong to, then ALTER COLUMN … SET NOT NULL. A fresh database never shows you this.

Platform accounts have no organization

organizationId is optional on the auth user, and the token exchange falls back to an empty string — not null — when the Shell's token carries no org. admin and system-admin are org-less platform roles (they seed with a blank organizationSlug), and admin@venizia.local is exactly the login Frontend suggests for the unprefixed /calendar. Pass that '' into a uuid NOT NULL column and the write dies with invalid input syntax for type uuid: "".

Check it before you touch the database, the way the platform's own apps do — reject with a 403 "Organization context required" rather than letting an empty string reach Postgres.

Done when — an event created under one org never appears when you're signed in under another.

Next: the last layer — ship it on its own pipeline. Ship →

Stuck? See Common snags, or back to Build a satellite.

Cross-system ​

Integration — invite people ​

Multi-tenancy — org calendars ​

Cross-system

Integration — invite people

Multi-tenancy — org calendars