diff --git a/api/src/db/drizzle/0002_tangled_repo.sql b/api/src/db/drizzle/0002_tangled_repo.sql new file mode 100644 index 0000000..d3c2a85 --- /dev/null +++ b/api/src/db/drizzle/0002_tangled_repo.sql @@ -0,0 +1,25 @@ +CREATE TABLE "tangled_repo" ( + "did" text NOT NULL, + "rev" text NOT NULL, + "rkey" text NOT NULL, + "cid" text, + "deleted" boolean NOT NULL, + "name" text, + "knot" text, + "spindle" text, + "description" text, + "website" text, + "topics" text[], + "source" text, + "labels" text[], + "repo_did" text, + "created_at" timestamp with time zone, + CONSTRAINT "tangled_repo_did_rkey_pk" PRIMARY KEY("did","rkey"), + CONSTRAINT "tangled_repo_cid_only_null_if_deleted" CHECK (("tangled_repo"."cid" is not null or "tangled_repo"."deleted" = true)), + CONSTRAINT "tangled_repo_non_null_fields" CHECK (("tangled_repo"."deleted" = true or ("tangled_repo"."name" is not null and "tangled_repo"."knot" is not null and "tangled_repo"."created_at" is not null))) +); +--> statement-breakpoint +ALTER TABLE "tangled_profile" ALTER COLUMN "cid" DROP NOT NULL;--> statement-breakpoint +ALTER TABLE "tangled_profile" ALTER COLUMN "bluesky" DROP NOT NULL;--> statement-breakpoint +ALTER TABLE "tangled_profile" ADD CONSTRAINT "tangled_profile_cid_only_null_if_deleted" CHECK (("tangled_profile"."cid" is not null or "tangled_profile"."deleted" = true));--> statement-breakpoint +ALTER TABLE "tangled_profile" ADD CONSTRAINT "tangled_profile_non_null_fields" CHECK (("tangled_profile"."deleted" = true or "tangled_profile"."bluesky" is not null)); \ No newline at end of file diff --git a/api/src/db/drizzle/meta/0002_snapshot.json b/api/src/db/drizzle/meta/0002_snapshot.json new file mode 100644 index 0000000..d13b2da --- /dev/null +++ b/api/src/db/drizzle/meta/0002_snapshot.json @@ -0,0 +1 @@ +{"id":"6bf06586-128f-4b67-9e6e-5f3ff761c8d3","prevId":"9f626a39-ec46-47c9-acba-9f815b1066cb","version":"7","dialect":"postgresql","tables":{"public.raw_records":{"name":"raw_records","schema":"","columns":{"id":{"name":"id","type":"uuidv7","primaryKey":true,"notNull":true},"tap_id":{"name":"tap_id","type":"bigint","primaryKey":false,"notNull":true},"received_at":{"name":"received_at","type":"timestamp with time zone","primaryKey":false,"notNull":true,"default":"now()"},"did":{"name":"did","type":"text","primaryKey":false,"notNull":true},"rev":{"name":"rev","type":"text","primaryKey":false,"notNull":true},"collection":{"name":"collection","type":"text","primaryKey":false,"notNull":true},"rkey":{"name":"rkey","type":"text","primaryKey":false,"notNull":true},"action":{"name":"action","type":"raw_action","typeSchema":"public","primaryKey":false,"notNull":true},"cid":{"name":"cid","type":"text","primaryKey":false,"notNull":false},"record":{"name":"record","type":"jsonb","primaryKey":false,"notNull":false}},"indexes":{"raw_records_collection_index":{"name":"raw_records_collection_index","columns":[{"expression":"collection","isExpression":false,"asc":true,"nulls":"last"}],"isUnique":false,"concurrently":false,"method":"btree","with":{}},"raw_records_did_collection_rkey_index":{"name":"raw_records_did_collection_rkey_index","columns":[{"expression":"did","isExpression":false,"asc":true,"nulls":"last"},{"expression":"collection","isExpression":false,"asc":true,"nulls":"last"},{"expression":"rkey","isExpression":false,"asc":true,"nulls":"last"}],"isUnique":false,"concurrently":false,"method":"btree","with":{}}},"foreignKeys":{},"compositePrimaryKeys":{},"uniqueConstraints":{"raw_records_tap_id_unique":{"name":"raw_records_tap_id_unique","nullsNotDistinct":false,"columns":["tap_id"]}},"policies":{},"checkConstraints":{"raw_records_not_null_unless_delete":{"name":"raw_records_not_null_unless_delete","value":"(\"raw_records\".\"action\" = 'delete' or (\"raw_records\".\"cid\" is not null and \"raw_records\".\"record\" is not null))"}},"isRLSEnabled":false},"public.tangled_profile":{"name":"tangled_profile","schema":"","columns":{"did":{"name":"did","type":"text","primaryKey":true,"notNull":true},"rev":{"name":"rev","type":"text","primaryKey":false,"notNull":true},"cid":{"name":"cid","type":"text","primaryKey":false,"notNull":false},"deleted":{"name":"deleted","type":"boolean","primaryKey":false,"notNull":true},"avatar":{"name":"avatar","type":"text","primaryKey":false,"notNull":false},"description":{"name":"description","type":"text","primaryKey":false,"notNull":false},"links":{"name":"links","type":"text[]","primaryKey":false,"notNull":false},"stats":{"name":"stats","type":"text[]","primaryKey":false,"notNull":false},"bluesky":{"name":"bluesky","type":"boolean","primaryKey":false,"notNull":false},"location":{"name":"location","type":"text","primaryKey":false,"notNull":false},"pinned_repositories":{"name":"pinned_repositories","type":"text[]","primaryKey":false,"notNull":false},"pronouns":{"name":"pronouns","type":"text","primaryKey":false,"notNull":false},"preferred_handle":{"name":"preferred_handle","type":"text","primaryKey":false,"notNull":false}},"indexes":{},"foreignKeys":{},"compositePrimaryKeys":{},"uniqueConstraints":{},"policies":{},"checkConstraints":{"tangled_profile_cid_only_null_if_deleted":{"name":"tangled_profile_cid_only_null_if_deleted","value":"(\"tangled_profile\".\"cid\" is not null or \"tangled_profile\".\"deleted\" = true)"},"tangled_profile_non_null_fields":{"name":"tangled_profile_non_null_fields","value":"(\"tangled_profile\".\"deleted\" = true or \"tangled_profile\".\"bluesky\" is not null)"}},"isRLSEnabled":false},"public.tangled_repo":{"name":"tangled_repo","schema":"","columns":{"did":{"name":"did","type":"text","primaryKey":false,"notNull":true},"rev":{"name":"rev","type":"text","primaryKey":false,"notNull":true},"rkey":{"name":"rkey","type":"text","primaryKey":false,"notNull":true},"cid":{"name":"cid","type":"text","primaryKey":false,"notNull":false},"deleted":{"name":"deleted","type":"boolean","primaryKey":false,"notNull":true},"name":{"name":"name","type":"text","primaryKey":false,"notNull":false},"knot":{"name":"knot","type":"text","primaryKey":false,"notNull":false},"spindle":{"name":"spindle","type":"text","primaryKey":false,"notNull":false},"description":{"name":"description","type":"text","primaryKey":false,"notNull":false},"website":{"name":"website","type":"text","primaryKey":false,"notNull":false},"topics":{"name":"topics","type":"text[]","primaryKey":false,"notNull":false},"source":{"name":"source","type":"text","primaryKey":false,"notNull":false},"labels":{"name":"labels","type":"text[]","primaryKey":false,"notNull":false},"repo_did":{"name":"repo_did","type":"text","primaryKey":false,"notNull":false},"created_at":{"name":"created_at","type":"timestamp with time zone","primaryKey":false,"notNull":false}},"indexes":{},"foreignKeys":{},"compositePrimaryKeys":{"tangled_repo_did_rkey_pk":{"name":"tangled_repo_did_rkey_pk","columns":["did","rkey"]}},"uniqueConstraints":{},"policies":{},"checkConstraints":{"tangled_repo_cid_only_null_if_deleted":{"name":"tangled_repo_cid_only_null_if_deleted","value":"(\"tangled_repo\".\"cid\" is not null or \"tangled_repo\".\"deleted\" = true)"},"tangled_repo_non_null_fields":{"name":"tangled_repo_non_null_fields","value":"(\"tangled_repo\".\"deleted\" = true or (\"tangled_repo\".\"name\" is not null and \"tangled_repo\".\"knot\" is not null and \"tangled_repo\".\"created_at\" is not null))"}},"isRLSEnabled":false}},"enums":{"public.raw_action":{"name":"raw_action","schema":"public","values":["create","update","delete"]}},"schemas":{},"sequences":{},"roles":{},"policies":{},"views":{},"_meta":{"columns":{},"schemas":{},"tables":{}}} diff --git a/api/src/db/drizzle/meta/_journal.json b/api/src/db/drizzle/meta/_journal.json index 2ae69cb..eacda96 100644 --- a/api/src/db/drizzle/meta/_journal.json +++ b/api/src/db/drizzle/meta/_journal.json @@ -15,6 +15,13 @@ "when": 1777773843186, "tag": "0001_tangled_profile", "breakpoints": true + }, + { + "idx": 2, + "version": "7", + "when": 1777821401885, + "tag": "0002_tangled_repo", + "breakpoints": true } ] } \ No newline at end of file diff --git a/api/src/db/tables/tangled.ts b/api/src/db/tables/tangled.ts index bf427e3..6b329ff 100644 --- a/api/src/db/tables/tangled.ts +++ b/api/src/db/tables/tangled.ts @@ -1,19 +1,62 @@ -import { boolean, pgTable, text } from "drizzle-orm/pg-core"; +import { and, eq, isNotNull, or, sql } from "drizzle-orm"; +import { boolean, check, pgTable, primaryKey, text, timestamp } from "drizzle-orm/pg-core"; export const tangledProfile = pgTable("tangled_profile", { did: text().primaryKey(), rev: text().notNull(), // no rkey here, because always self - cid: text().notNull(), + cid: text(), deleted: boolean().notNull(), avatar: text(), description: text(), links: text().array(), stats: text().array(), - bluesky: boolean().notNull(), + bluesky: boolean(), location: text(), pinnedRepositories: text().array(), pronouns: text(), preferredHandle: text(), -}); +}, (table) => [ + check("tangled_profile_cid_only_null_if_deleted", or(isNotNull(table.cid), eq(table.deleted, sql`true`))!), + check("tangled_profile_non_null_fields", + or( + eq(table.deleted, sql`true`), + and( + isNotNull(table.bluesky) + ) + )! + ) +]); + +export const tangledRepo = pgTable("tangled_repo", { + did: text().notNull(), + rev: text().notNull(), + rkey: text().notNull(), + cid: text(), + deleted: boolean().notNull(), + + name: text(), + knot: text(), + spindle: text(), + description: text(), + website: text(), + topics: text().array(), + source: text(), + labels: text().array(), + repoDid: text(), + createdAt: timestamp({ withTimezone: true }), +}, (table) => [ + primaryKey({ columns: [table.did, table.rkey] }), + check("tangled_repo_cid_only_null_if_deleted", or(isNotNull(table.cid), eq(table.deleted, sql`true`))!), + check("tangled_repo_non_null_fields", + or( + eq(table.deleted, sql`true`), + and( + isNotNull(table.name), + isNotNull(table.knot), + isNotNull(table.createdAt), + ) + )! + ) +]); diff --git a/api/src/ingest/profile.ts b/api/src/ingest/profile.ts index 69bcdaa..f8c5c89 100644 --- a/api/src/ingest/profile.ts +++ b/api/src/ingest/profile.ts @@ -7,17 +7,28 @@ import { tangledProfile } from "../db/tables/tangled.ts"; export const ingestProfile: CollectionHandler = async (profiles) => { for (const profile of profiles) { - const validatedRecord = lexiconProfile.$safeParse(profile.record); - if (!validatedRecord.success) continue; - const { value: profileRecord } = validatedRecord; - if (profile.action === "delete") { - await db.update(tangledProfile).set({ - deleted: true - }).where(and(eq(tangledProfile.did, profile.did), lt(tangledProfile.rev, profile.rev))); + await db.insert(tangledProfile).values({ + did: profile.did, + rev: profile.rev, + cid: null, + deleted: true, + }).onConflictDoUpdate({ + target: tangledProfile.did, + set: { + rev: profile.rev, + cid: null, + deleted: true + }, + setWhere: lt(tangledProfile.rev, profile.rev) + }); continue; } + const validatedRecord = lexiconProfile.$safeParse(profile.record); + if (!validatedRecord.success) continue; + const { value: profileRecord } = validatedRecord; + const avatar = profileRecord.avatar ? getBlobCidString(profileRecord.avatar) : null; await db.insert(tangledProfile).values({ @@ -51,7 +62,8 @@ export const ingestProfile: CollectionHandler = async (profiles) => { pinnedRepositories: profileRecord.pinnedRepositories, pronouns: profileRecord.pronouns, preferredHandle: profileRecord.preferredHandle, - } + }, + setWhere: lt(tangledProfile.rev, profile.rev) }) } } diff --git a/api/src/ingest/repo.ts b/api/src/ingest/repo.ts new file mode 100644 index 0000000..29e86c7 --- /dev/null +++ b/api/src/ingest/repo.ts @@ -0,0 +1,71 @@ +import { and, eq, lt } from "drizzle-orm"; +import { getBlobCidString } from "@atproto/lex"; +import type { CollectionHandler } from "./tap"; +import * as lexiconRepo from "../lexicons/sh/tangled/repo.ts"; +import { db } from "../db/index.ts"; +import { tangledRepo } from "../db/tables/tangled.ts"; + +export const ingestRepo: CollectionHandler = async (repos) => { + for (const repo of repos) { + if (repo.action === "delete") { + await db.insert(tangledRepo).values({ + did: repo.did, + rev: repo.rev, + rkey: repo.rkey, + cid: null, + deleted: true, + }).onConflictDoUpdate({ + target: [tangledRepo.did, tangledRepo.rkey], + set: { + rev: repo.rev, + cid: null, + deleted: true + }, + setWhere: lt(tangledRepo.rev, repo.rev) + }); + continue; + } + + const validatedRecord = lexiconRepo.$safeParse(repo.record); + if (!validatedRecord.success) continue; + const { value: repoRecord } = validatedRecord; + + await db.insert(tangledRepo).values({ + did: repo.did, + rev: repo.rev, + rkey: repo.rkey, + cid: repo.cid, + deleted: false, + + name: repoRecord.name, + knot: repoRecord.knot, + spindle: repoRecord.spindle, + description: repoRecord.description, + website: repoRecord.website, + topics: repoRecord.topics, + source: repoRecord.source, + labels: repoRecord.labels, + repoDid: repoRecord.repoDid, + createdAt: new Date(repoRecord.createdAt), + }).onConflictDoUpdate({ + target: [tangledRepo.did, tangledRepo.rkey], + set: { + rev: repo.rev, + cid: repo.cid, + deleted: false, + + name: repoRecord.name, + knot: repoRecord.knot, + spindle: repoRecord.spindle, + description: repoRecord.description, + website: repoRecord.website, + topics: repoRecord.topics, + source: repoRecord.source, + labels: repoRecord.labels, + repoDid: repoRecord.repoDid, + createdAt: new Date(repoRecord.createdAt), + }, + setWhere: lt(tangledRepo.rev, repo.rev) + }) + } +} diff --git a/api/src/ingest/tap.ts b/api/src/ingest/tap.ts index 58da62e..f17df36 100644 --- a/api/src/ingest/tap.ts +++ b/api/src/ingest/tap.ts @@ -4,6 +4,7 @@ import { db } from '../db/index.ts'; import { rawRecords } from '../db/tables/raw_records.ts'; import { TAP_URL } from '../lib/constants.ts'; import { ingestProfile } from './profile.ts'; +import { ingestRepo } from './repo.ts'; const tap = new Tap(TAP_URL); @@ -35,7 +36,8 @@ type CollectionHandlerArg = { export type CollectionHandler = (updates: CollectionHandlerArg[]) => Promise; const COLLECTION_HANDLERS: Record = { - "sh.tangled.actor.profile": ingestProfile + "sh.tangled.actor.profile": ingestProfile, + "sh.tangled.repo": ingestRepo, }; async function syncLevelTwoTables(insertedRecords: CollectionHandlerArg[], log = false) {