`@octabits-io/activity`

The in-app notification feed over @octabits-io/events: the bell, its badge, read state, and the rule that makes both honest. One row per fact, written by the same call that emits the event, in the same transaction. Read state per user in a side table. The audience is evaluated at read time against the reader’s live grants, never snapshotted. Resolution marks the work done for everyone. Per-user mutes, a capped unread count, pruning. A store contract with an in-memory reference, a plain-SQL Postgres adapter over any executor, a Drizzle adapter, and a framework-free feed store for the browser with Vue and React bindings.

pnpm add @octabits-io/activity @octabits-io/events

Subpath	Deps	Contents
`.`	`@octabits-io/events` (vocabulary + the permission predicate), `@octabits-io/result`	the `ActivityStore` contract, the in-memory store, `createActivityFeed` (read side), `createActivityPublisher` (write side)
`./postgres`	none	plain SQL over a structural `SqlExecutor` — `pg` Pool/Client, PGlite, a wrapper over postgres.js — plus `activityDdl()`
`./drizzle`	`drizzle-orm` (optional peer), `@octabits-io/postgres` (peer)	`activityColumns`, `activityReadColumns`, `activityMuteColumns` to spread into your tables (timestamps read back as ISO-8601 in UTC); `createDrizzleActivityStore` (`tables.mute` optional — without it there are no per-user mutes)
`./client`	none	`createActivityFeedStore` — the bell, headless: pages, badge, mark read, live frames in
`./vue`, `./react`	`vue` / `react` (optional peers)	`useActivityFeed(store)`

What is generic, and what stays yours

The generic half is the shape every bell has: a row per fact, who has read it, who may see it, when the work behind it was done, what a person muted, and how old is too old. That is this package.

The half that stays in your app is the vocabulary: the event types, the i18n keys, the deep links, which events deserve action_required, and which permission each type needs. Rows store a title key and params, never a rendered sentence — a person can switch locale after the row was written — and never personal data: the table is plaintext.

When not to use it

Without the event lane. The write side is emit on an events publisher: the row and the durable event are one transaction, and the bell is live because the frame carries the row. A feed written without @octabits-io/events is a table you can build without this package.
An audit trail. Rows are pruned, muted, resolved and hidden by audience; they carry a title key and params, never the change. What an agent did and how to undo it is @octabits-io/agent-ledger; what a person did is your own audit table.
Reaching people who are not in the app. In-app only: no e-mail, no push, no digest, no device. A notification that must leave the browser is a mail sent from your own trigger — the same emit, a second sink.
Messages. A row is a fact about a record, not a message from a person: no body, no thread, no reply. An inbox is a different table.
Work with owners and deadlines. resolve marks a fact handled for everyone; there is no assignment, no due date, no state beyond open and resolved. A task is a record of yours; the feed can point at it.
Anything personal in params. The table is plaintext and read by everyone the audience admits. Put ids in the row and resolve them at render, or encrypt with @octabits-io/pii before it gets here.

Writing

import { createActivityPublisher } from '@octabits-io/activity';
import { createDrizzleActivityStore } from '@octabits-io/activity/drizzle';

const store = createDrizzleActivityStore({ db, tables: { activity, read, mute }, scope: { column: 'orgId', value: orgId } });
const activity = createActivityPublisher({ store, publisher: eventPublisher, scopeKey: orgId });

await db.transaction(async (tx) => {
  const contact = await contacts.create(input, tx);
  // The feed row and the durable outbox event, on the contact's own transaction.
  await activity.emit(
    {
      type: 'contact.created',
      data: { contactId: contact.id },
      actor: { type: 'user', id: userId },
      resources: [`contact:${contact.id}`],
      audience: { permission: { contacts: ['read'] } },
      activity: { titleKey: 'activity.contact.created', params: { contactId: contact.id }, link: `/contacts/${contact.id}`, severity: 'info' },
    },
    tx,
  );
});

// Later, once the work behind some activities was handled — after the commit, never inside it:
await activity.resolve({ resources: [`draft:${draftId}`], types: ['draft.created'] });

emit throws on failure: inside the transaction the throw is the rollback, and a state change whose activity cannot be recorded must not half-succeed. The event carries the row inline as data.activity, so a connected client renders it without a refetch.

resolve is best-effort by design: it stamps resolvedAt on the unresolved rows whose resources overlap, tells connected clients which ids settled (an ephemeral activity.resolved event — resolvedEvent: { type, audience } names your own type when your publisher validates types against a registry), logs failures and never throws — bell hygiene must not fail real work.

Reading

import { createActivityFeed } from '@octabits-io/activity';

const feed = createActivityFeed({
  store,
  scopeKey: orgId,
  reader: { userId, can: (permission) => rbac.allows(role, permission) },
  badgeSeverities: ['action_required'], // optional: the badge counts open work only; `info` rows light the list
});

const page = await feed.list({ limit: 20 });          // { items, nextBefore }
const needed = await feed.list({ since, severities: ['action_required'], types: ['order.requested'] }); // predicates, for an assistant
const history = await feed.list({ resources: ['order:42'] }); // one record's history: rows whose resources overlap, still gated per reader
const unread = await feed.unreadCount();              // { count, capped } — stops at 99
await feed.markRead([page.value.items[0].id]);           // { marked: 1 }
await feed.markAllRead();                                // { marked: n }
await feed.setMutedTypes(['import.progress']);

Two rules, applied everywhere:

audienceAllows — a row with audience.users is for those users; a row with audience.permission is for readers whose can(permission) says so, and a reader with no can sees none of those (fail closed, as the hub does). Evaluated per row, per read, so a grant revoked yesterday hides yesterday’s rows too.
isReadFor — a row is read for a reader when they marked it, when it was resolved, or when they caused it. Nobody needs a badge for their own doing.

list over-fetches: the audience gate runs in the application, so a page of limit visible rows may need more raw rows. The unread count asks the store for cap + 1 candidates, gates them, and stops at the cap. markAllRead walks the reader’s unread rows the same way and marks only those the gate lets them see (muted types included): a row for someone else keeps no marker of theirs, so a grant given later shows it unread.

The bell

import { createActivityFeedStore } from '@octabits-io/activity/client';
import { useActivityFeed } from '@octabits-io/activity/vue'; // or /react

const store = createActivityFeedStore({
  userId,
  api: {
    list: (params) => fetchJson('/api/activity', params),
    unreadCount: () => fetchJson('/api/activity/unread-count'),
    markRead: (ids) => postJson('/api/activity/read', { ids }),
    markAllRead: () => postJson('/api/activity/read-all'),
  },
});
eventStream.onEvent = (event) => store.applyEvent(event); // the live half

const feed = useActivityFeed(store); // entries, unread, loading, loaded, exhausted, error + the actions

applyEvent takes what the event stream delivers: a durable frame carrying data.activity is prepended (and counted, unless this user caused it); an activity.resolved frame settles the rows it names and refetches the count. Marking read is optimistic; a failed call restores the count from the server, and so does any read while the badge is capped.

Give the store the same badgeSeverities as the server feed. With ['action_required'] the server’s count leaves info rows out, and a store left on the default would count them as they arrive and subtract them as they are read, so the badge would drift from the server’s until the next refetch.

Scoping

scope: { column, value } binds an adapter to one scope: every row stamped, every read filtered. scope: { column } stamps each row with its own scopeKey and reads by it — for a process serving many scopes. No scope at all is the single-scope app, and the adapters refuse a second scopeKey rather than mix. activityDdl({ scope: { column: 'org_id' } }) emits the column on all three tables; row-level security is yours to add — the column is there for it.

Conformance

src/conformance.test.ts runs the in-memory, plain-SQL and Drizzle stores through one scenario on an embedded PGlite. A fourth adapter passes the same suite or is not an adapter.

Wire format

The feed’s shapes are published as JSON Schema (draft 2020-12):

Document	What
`schema/activity-entry.schema.json`	`ActivityEntry` — a row as the reader sees it; `ActivityFeedItem` (the slice that rides inline on the event as `data.activity`) is its `$defs` entry
`schema/activity-page.schema.json`	`ActivityPage` — items newest first, `nextBefore` as the cursor
`schema/unread-count.schema.json`	`UnreadCount` — the badge, with `capped`

The runtime is plain interfaces and takes no Zod; the documents come from a Zod mirror in scripts/wire.ts, and src/wire.test.ts proves the mirror equals the interfaces at the type level, so the published schema cannot drift from what the store returns. Checked in pnpm lint; in the package (@octabits-io/activity/schema/activity-entry.schema.json) and at their $id, the raw GitHub URL.

License

MIT