takeEvent

tcmts.registry.takeEvent · #phlfh7tp · term · runs on the server

uses tcmts.registry.releases, which uses tcmts.server.table, which uses tcmts.server.Table, which uses #tcm/host (database)

One Jetstream event, into the tables: a release record announced (a new row, not shown until it's checked and mirrored), changed or deleted (checked again against its author's repository first, recheckRelease: an event alone changes nothing the registry has checked); or a review, advisory, yank or star (its check queued), or one deleted.

import { ATTESTATION_KINDS, recordKey, RELEASE_COLLECTION, STAR_COLLECTION } from "#codebase/tcmts/records";
import {
  mirror, people, recheckRelease, recordOf, releases, takeAttestation, takeStar,
} from "#codebase/tcmts/registry";
import type { RegistryRelease } from "#codebase/tcmts/registry";
import { enqueue } from "#codebase/tcmts/server";
import type { RawEvent } from "@bsky/jetstream";

export async function takeEvent(e: RawEvent): Promise<void> {
  if (e.kind !== "commit") return;
  if (Object.values(ATTESTATION_KINDS).includes(e.commit.collection as never))
    return takeAttestation(e);
  if (e.commit.collection === STAR_COLLECTION) return takeStar(e);
  if (e.commit.collection !== RELEASE_COLLECTION) return;
  const uri = `at://${e.did}/${RELEASE_COLLECTION}/${e.commit.rkey}`;
  const had = await releases.findOne({ uri });
  if (!(await people.findOne({ did: e.did }))) await people.insert({ did: e.did, seq: e.seq });
  if (e.commit.operation === "delete") {
    if (had) enqueue(recheckRelease, uri);
    return;
  }
  const r = recordOf(e.commit.record);
  if (!r || recordKey(r) !== e.commit.rkey) return;
  if (had) {
    enqueue(recheckRelease, uri);
    return;
  }
  const row: RegistryRelease = {
    uri,
    did: e.did,
    rkey: e.commit.rkey,
    cid: e.commit.cid,
    package: r.package,
    version: r.version,
    namespace: r.namespace,
    entry: r.entry,
    notes: r.notes,
    sources: r.sources,
    createdAt: r.createdAt,
    seq: e.seq,
    state: "pending",
  };
  await releases.insert(row);
  enqueue(mirror, uri);
}

uses

used by