tapping in

.gitignore

···
 vite.config.ts.timestamp-*
 
 .idea
-local.db
+local.db
+
+#tap
+tap.db
+tap.db-shm
+tap.db-wal

dev.compose.yml

···
       timeout: 5s
     networks:
       - services-network
+  tap:
+    image: ghcr.io/bluesky-social/indigo/tap:latest
+    platform: linux/amd64
+    depends_on:
+      - postgres
+      - valkey
+    ports:
+      - '2480:2480'
+    env_file:
+      - .env
+    extra_hosts:
+      - "host.docker.internal:host-gateway"
+    network_mode: "host"
+
 volumes:
   valkey_data:
   postgres_data:

drizzle/0000_breezy_menace.sql

···
+CREATE TABLE "record_pokes" (
+	"id" serial PRIMARY KEY NOT NULL,
+	"recordId" integer,
+	"pokersRepo" text NOT NULL,
+	"atUri" text NOT NULL,
+	"indexedAt" time DEFAULT now() NOT NULL
+);
+--> statement-breakpoint
+CREATE TABLE "records" (
+	"id" serial PRIMARY KEY NOT NULL,
+	"rkey" varchar NOT NULL,
+	"collection" varchar NOT NULL,
+	"repo" varchar NOT NULL,
+	"atUri" text NOT NULL,
+	"data" jsonb NOT NULL,
+	"indexedAt" timestamp DEFAULT now() NOT NULL
+);
+--> statement-breakpoint
+CREATE TABLE "user_pokes" (
+	"id" serial PRIMARY KEY NOT NULL,
+	"subject" text NOT NULL,
+	"poker" text NOT NULL,
+	"at_uri" text NOT NULL,
+	"indexedAt" time DEFAULT now() NOT NULL
+);
+--> statement-breakpoint
+ALTER TABLE "record_pokes" ADD CONSTRAINT "record_pokes_recordId_records_id_fk" FOREIGN KEY ("recordId") REFERENCES "public"."records"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
+CREATE INDEX "record_pokes_pokersRepo_index" ON "record_pokes" USING btree ("pokersRepo");--> statement-breakpoint
+CREATE INDEX "record_pokes_atUri_index" ON "record_pokes" USING btree ("atUri");--> statement-breakpoint
+CREATE INDEX "records_rkey_index" ON "records" USING btree ("rkey");--> statement-breakpoint
+CREATE INDEX "records_collection_index" ON "records" USING btree ("collection");--> statement-breakpoint
+CREATE INDEX "records_repo_index" ON "records" USING btree ("repo");--> statement-breakpoint
+CREATE UNIQUE INDEX "records_atUri_index" ON "records" USING btree ("atUri");--> statement-breakpoint
+CREATE INDEX "user_pokes_subject_index" ON "user_pokes" USING btree ("subject");--> statement-breakpoint
+CREATE INDEX "user_pokes_poker_index" ON "user_pokes" USING btree ("poker");

drizzle/0000_yielding_psylocke.sql

···
-CREATE TABLE "record_pokes" (
-	"id" serial PRIMARY KEY NOT NULL,
-	"recordId" integer,
-	"pokersRepo" text NOT NULL,
-	"atUri" text NOT NULL,
-	"indexedAt" time DEFAULT now() NOT NULL
-);
---> statement-breakpoint
-CREATE TABLE "records" (
-	"id" serial PRIMARY KEY NOT NULL,
-	"rkey" varchar NOT NULL,
-	"collection" varchar NOT NULL,
-	"repo" varchar NOT NULL,
-	"data" jsonb NOT NULL,
-	"indexedAt" time DEFAULT now() NOT NULL
-);
---> statement-breakpoint
-CREATE TABLE "user_pokes" (
-	"id" serial PRIMARY KEY NOT NULL,
-	"subject" text NOT NULL,
-	"poker" text NOT NULL,
-	"at_uri" text NOT NULL,
-	"indexedAt" time DEFAULT now() NOT NULL
-);
---> statement-breakpoint
-ALTER TABLE "record_pokes" ADD CONSTRAINT "record_pokes_recordId_records_id_fk" FOREIGN KEY ("recordId") REFERENCES "public"."records"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
-CREATE INDEX "pokers_repo_idx" ON "record_pokes" USING btree ("pokersRepo");--> statement-breakpoint
-CREATE INDEX "at_uri_idx" ON "record_pokes" USING btree ("atUri");--> statement-breakpoint
-CREATE INDEX "rkey_idx" ON "records" USING btree ("rkey");--> statement-breakpoint
-CREATE INDEX "collection_idx" ON "records" USING btree ("collection");--> statement-breakpoint
-CREATE INDEX "repo_idx" ON "records" USING btree ("repo");--> statement-breakpoint
-CREATE INDEX "subject_idx" ON "user_pokes" USING btree ("subject");--> statement-breakpoint
-CREATE INDEX "poker_idx" ON "user_pokes" USING btree ("poker");

drizzle/meta/0000_snapshot.json

···
 {
-  "id": "eb652b7e-c43a-411e-9e94-15bfb4be773f",
+  "id": "b5b063c6-2451-42ca-8bef-0119b8202cb3",
   "prevId": "00000000-0000-0000-0000-000000000000",
   "version": "7",
   "dialect": "postgresql",
···
         }
       },
       "indexes": {
-        "pokers_repo_idx": {
-          "name": "pokers_repo_idx",
+        "record_pokes_pokersRepo_index": {
+          "name": "record_pokes_pokersRepo_index",
           "columns": [
             {
               "expression": "pokersRepo",
···
           "method": "btree",
           "with": {}
         },
-        "at_uri_idx": {
-          "name": "at_uri_idx",
+        "record_pokes_atUri_index": {
+          "name": "record_pokes_atUri_index",
           "columns": [
             {
               "expression": "atUri",
···
           "primaryKey": false,
           "notNull": true
         },
+        "atUri": {
+          "name": "atUri",
+          "type": "text",
+          "primaryKey": false,
+          "notNull": true
+        },
         "data": {
           "name": "data",
           "type": "jsonb",
···
         },
         "indexedAt": {
           "name": "indexedAt",
-          "type": "time",
+          "type": "timestamp",
           "primaryKey": false,
           "notNull": true,
           "default": "now()"
         }
       },
       "indexes": {
-        "rkey_idx": {
-          "name": "rkey_idx",
+        "records_rkey_index": {
+          "name": "records_rkey_index",
           "columns": [
             {
               "expression": "rkey",
···
           "method": "btree",
           "with": {}
         },
-        "collection_idx": {
-          "name": "collection_idx",
+        "records_collection_index": {
+          "name": "records_collection_index",
           "columns": [
             {
               "expression": "collection",
···
           "method": "btree",
           "with": {}
         },
-        "repo_idx": {
-          "name": "repo_idx",
+        "records_repo_index": {
+          "name": "records_repo_index",
           "columns": [
             {
               "expression": "repo",
···
             }
           ],
           "isUnique": false,
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        },
+        "records_atUri_index": {
+          "name": "records_atUri_index",
+          "columns": [
+            {
+              "expression": "atUri",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": true,
           "concurrently": false,
           "method": "btree",
           "with": {}
···
         }
       },
       "indexes": {
-        "subject_idx": {
-          "name": "subject_idx",
+        "user_pokes_subject_index": {
+          "name": "user_pokes_subject_index",
           "columns": [
             {
               "expression": "subject",
···
           "method": "btree",
           "with": {}
         },
-        "poker_idx": {
-          "name": "poker_idx",
+        "user_pokes_poker_index": {
+          "name": "user_pokes_poker_index",
           "columns": [
             {
               "expression": "poker",

drizzle/meta/_journal.json

···
     {
       "idx": 0,
       "version": "7",
-      "when": 1766032962283,
-      "tag": "0000_yielding_psylocke",
+      "when": 1766121859810,
+      "tag": "0000_breezy_menace",
       "breakpoints": true
     }
   ]

package.json

···
 		"@atproto/jwk-jose": "^0.1.11",
 		"@atproto/oauth-client-node": "^0.3.13",
 		"@atproto/oauth-types": "^0.5.2",
+		"@atproto/tap": "^0.0.2",
 		"@oslojs/crypto": "^1.0.1",
 		"@oslojs/encoding": "^1.1.0",
 		"@sveltejs/adapter-node": "^5.4.0",

pnpm-lock.yaml

···
       '@atproto/oauth-types':
         specifier: ^0.5.2
         version: 0.5.2
+      '@atproto/tap':
+        specifier: ^0.0.2
+        version: 0.0.2
       '@oslojs/crypto':
         specifier: ^1.0.1
         version: 1.0.1
···
 
   '@atproto/syntax@0.4.2':
     resolution: {integrity: sha512-X9XSRPinBy/0VQ677j8VXlBsYSsUXaiqxWVpGGxJYsAhugdQRb0jqaVKJFtm6RskeNkV6y9xclSUi9UYG/COrA==}
+
+  '@atproto/tap@0.0.2':
+    resolution: {integrity: sha512-CrfJWrvozuSIokOQLMeSFcF5ZpstpxIZ9PnBpgIkbLQQKb3wO+0dn90xZN5jlLjczPHvT4PrF1z8uYgVlujTlg==}
+    engines: {node: '>=18.7.0'}
+
+  '@atproto/ws-client@0.0.4':
+    resolution: {integrity: sha512-dox1XIymuC7/ZRhUqKezIGgooZS45C6vHCfu0PnWjfvsLCK2kAlnvX4IBkA/WpcoijDhQ9ejChnFbo/sLmgvAg==}
+    engines: {node: '>=18.7.0'}
 
   '@atproto/xrpc@0.7.7':
     resolution: {integrity: sha512-K1ZyO/BU8JNtXX5dmPp7b5UrkLMMqpsIa/Lrj5D3Su+j1Xwq1m6QJ2XJ1AgjEjkI1v4Muzm7klianLE6XGxtmA==}
···
   wrappy@1.0.2:
     resolution: {integrity: sha512-l4Sp/DRseor9wL6EvV2+TuQn63dMkPjZ/sp9XkghTEbV9KlPS1xUsZ3u7/IQO4wxtcFB4bgpQPRcR3QCvezPcQ==}
 
+  ws@8.18.3:
+    resolution: {integrity: sha512-PEIGCY5tSlUt50cqyMXfCzX+oOPqN0vuGqWzbcJ2xvnkzkq46oOpz7dQaTDBdfICb4N14+GARUDw2XV2N4tvzg==}
+    engines: {node: '>=10.0.0'}
+    peerDependencies:
+      bufferutil: ^4.0.1
+      utf-8-validate: '>=5.0.2'
+    peerDependenciesMeta:
+      bufferutil:
+        optional: true
+      utf-8-validate:
+        optional: true
+
   xtend@4.0.2:
     resolution: {integrity: sha512-LKYU1iAXJXUgAXn9URjiu+MWhyUXHsvfp7mcuYm9dSUKK0/CjtrUwFAxD82/mCWbtLsGjFIad0wIsod4zrTAEQ==}
     engines: {node: '>=0.4'}
···
       zod: 3.25.76
 
   '@atproto/syntax@0.4.2': {}
+
+  '@atproto/tap@0.0.2':
+    dependencies:
+      '@atproto/common': 0.5.3
+      '@atproto/syntax': 0.4.2
+      '@atproto/ws-client': 0.0.4
+      ws: 8.18.3
+      zod: 3.25.76
+    transitivePeerDependencies:
+      - bufferutil
+      - utf-8-validate
+
+  '@atproto/ws-client@0.0.4':
+    dependencies:
+      '@atproto/common': 0.5.3
+      ws: 8.18.3
+    transitivePeerDependencies:
+      - bufferutil
+      - utf-8-validate
 
   '@atproto/xrpc@0.7.7':
     dependencies:
···
 
   wrappy@1.0.2:
     optional: true
+
+  ws@8.18.3: {}
 
   xtend@4.0.2: {}
 

src/lib/server/db/index.ts

···
 if (!env.DATABASE_URL) throw new Error('DATABASE_URL is not set');
 
 logger.info('Connected to database');
-logger.info(`Database url: ${env.DATABASE_URL}`);
 export const db = drizzle(env.DATABASE_URL!,{ schema });

src/lib/server/db/schema.ts

···
-import { index, text, integer, jsonb, pgTable, serial, time, varchar } from 'drizzle-orm/pg-core';
+import { index, text, integer, jsonb, pgTable, serial, time, varchar, uniqueIndex, timestamp } from 'drizzle-orm/pg-core';
 
 /**
  * Holds collected records from firehose, backfilled, or manually entered
···
 	rkey: varchar().notNull(),
 	collection: varchar().notNull(),
 	repo: varchar().notNull(),
+	atUri: text().notNull(),
 	data: jsonb().notNull(),
-	indexedAt: time().notNull().defaultNow()
+	indexedAt: timestamp({ mode: 'date' }).notNull().defaultNow()
 }, (table) => [
-	index('rkey_idx').on(table.rkey),
-	index('collection_idx').on(table.collection),
-	index('repo_idx').on(table.repo),
+	index().on(table.rkey),
+	index().on(table.collection),
+	index().on(table.repo),
+	uniqueIndex().on(table.atUri),
 ]);
 
 
···
 	atUri: text().notNull(),
 	indexedAt: time().notNull().defaultNow()
 },(table) => [
-	index('pokers_repo_idx').on(table.pokersRepo),
-	index('at_uri_idx').on(table.atUri)
+	index().on(table.pokersRepo),
+	index().on(table.atUri)
 ]);
 
 /**
···
 	at_uri: text().notNull(),
 	indexedAt: time().notNull().defaultNow()
 }, (table) => [
-	index('subject_idx').on(table.subject),
-	index('poker_idx').on(table.poker)
+	index().on(table.subject),
+	index().on(table.poker)
 ]);
 

src/routes/webhooks/tap/+server.ts

···
+import { json } from '@sveltejs/kit';
+import { env } from '$env/dynamic/private';
+import type { RequestHandler } from './$types';
+import { logger } from '$lib/server/logger';
+import { assureAdminAuth, type RecordEvent, type TapEvent } from '@atproto/tap';
+import { db } from '$lib/server/db';
+import { recordsTable } from '$lib/server/db/schema';
+import { eq } from 'drizzle-orm';
+
+export const POST: RequestHandler = async ({ request }) => {
+    const auth = request.headers.get('Authorization');
+
+    if (auth === null) {
+        logger.error('Missing Authorization header');
+        return json({ error: 'Missing Authorization header' }, { status: 401 });
+    }
+
+    try{
+        if(env.TAP_ADMIN_PASSWORD === undefined) throw new Error('TAP_ADMIN_PASSWORD is not set');
+        assureAdminAuth(env.TAP_ADMIN_PASSWORD, auth);
+    }catch (err){
+        const errorMessage = (err as Error).message;
+        logger.error('Tap webhook auth error: ' + errorMessage + '');
+        return json({ error: 'Not authenticated' }, { status: 401 });
+    }
+
+    try {
+        const body = await request.json();
+        await parseAndProcessTapEvent(body);
+        
+
+        //This should just respond with 200 OK. comment out to not ack
+        return json({  });
+    } catch (err) {
+        console.error('Failed to process event:', err);
+        return json({ error: 'Failed to process event' }, { status: 500 });
+    }
+
+};
+
+const parseAndProcessTapEvent = async (event: TapEvent) => {
+    switch (event.type) {
+        case 'identity':
+            logger.info(event);
+            break;
+        case 'record':
+            await saveRecord(event.record as RecordEvent);
+            break;
+        default:
+            throw new Error(`Unsupported event type: ${JSON.stringify(event)}`);
+    }
+};
+
+
+const saveRecord = async (event: RecordEvent) => {
+    const atUri = `${event.did}/${event.collection}/${event.rkey}`;
+    logger.info(`Processing record event: ${atUri}`);
+    if (event.action === 'create' || event.action === 'update') {
+        if (!event.record) {
+            logger.warn(`Record event with action ${event.action} missing record data: ${event.rkey}`);
+            return;
+        }
+
+        await db.insert(recordsTable).values({
+            rkey: event.rkey,
+            collection: event.collection,
+            repo: event.did,
+            atUri: atUri,
+            data: event.record,
+        }).onConflictDoUpdate({
+            target: recordsTable.atUri,
+            set: {
+                data: event.record,
+
+            }
+        });
+        logger.info(`Saved record: ${event.did}/${event.collection}/${event.rkey}`);
+    } else if (event.action === 'delete') {
+        await db.delete(recordsTable).where(eq(recordsTable.atUri, atUri));
+    }
+};