diff --git a/actions/getIdentityData.ts b/actions/getIdentityData.ts index 08684ce1..754c3b70 100644 --- a/actions/getIdentityData.ts +++ b/actions/getIdentityData.ts @@ -15,6 +15,7 @@ export async function getIdentityData() { `*, identities( *, + bsky_profiles(*), subscribers_to_publications(*), custom_domains(*), home_leaflet:permission_tokens!identities_home_page_fkey(*, permission_token_rights(*)), diff --git a/app/lish/subscribeToPublication.ts b/app/lish/subscribeToPublication.ts index 314daaa8..73f7deaf 100644 --- a/app/lish/subscribeToPublication.ts +++ b/app/lish/subscribeToPublication.ts @@ -10,6 +10,7 @@ import { revalidatePath } from "next/cache"; import { AtUri } from "@atproto/syntax"; import { redirect } from "next/navigation"; import { encodeActionToSearchParam } from "app/api/oauth/[route]/afterSignInActions"; +import { Json } from "supabase/database.types"; let leafletFeedURI = "at://did:plc:btxrwcaeyodrap5mnjw2fvmz/app.bsky.feed.generator/subscribedPublications"; @@ -44,7 +45,19 @@ export async function subscribeToPublication( identity: credentialSession.did!, }); let bsky = new BskyAgent(credentialSession); - let prefs = await bsky.app.bsky.actor.getPreferences(); + let [prefs, profile] = await Promise.all([ + bsky.app.bsky.actor.getPreferences(), + bsky.app.bsky.actor.profile.get({ + repo: credentialSession.did!, + rkey: "self", + }), + ]); + if (!identity.bsky_profiles && profile.value) { + await supabaseServerClient.from("bsky_profiles").insert({ + did: identity.atp_did, + record: profile.value as Json, + }); + } let savedFeeds = prefs.data.preferences.find( (pref) => pref.$type === "app.bsky.actor.defs#savedFeedsPrefV2", ) as AppBskyActorDefs.SavedFeedsPrefV2; diff --git a/appview/index.ts b/appview/index.ts index df5a3d63..2e52bef8 100644 --- a/appview/index.ts +++ b/appview/index.ts @@ -11,6 +11,10 @@ import { } from "lexicons/api"; import { AtUri } from "@atproto/syntax"; import { writeFile, readFile } from "fs/promises"; +import { createIdentity } from "actions/createIdentity"; +import { supabaseServerClient } from "supabase/serverClient"; +import postgres from "postgres"; +import { drizzle } from "drizzle-orm/postgres-js"; const cursorFile = process.env.CURSOR_FILE || "/cursor/cursor"; @@ -23,6 +27,9 @@ async function main() { try { startCursor = parseInt((await readFile(cursorFile)).toString()); } catch (e) {} + + const client = postgres(process.env.DB_URL!); + const db = drizzle(client); const runner = new MemoryRunner({ startCursor, setCursor: async (cursor) => { @@ -40,6 +47,7 @@ async function main() { ids.PubLeafletDocument, ids.PubLeafletPublication, ids.PubLeafletGraphSubscription, + ids.AppBskyActorProfile, ], handleEvent: async (evt) => { if ( @@ -81,12 +89,22 @@ async function main() { if (evt.event === "create" || evt.event === "update") { let record = PubLeafletPublication.validateRecord(evt.record); if (!record.success) return; - await supabase.from("publications").upsert({ + let { error } = await supabase.from("publications").upsert({ uri: evt.uri.toString(), identity_did: evt.did, name: record.value.name, record: record.value as Json, }); + + if (error && error.code === "23503") { + await createIdentity(db, { atp_did: evt.did }); + await supabase.from("publications").upsert({ + uri: evt.uri.toString(), + identity_did: evt.did, + name: record.value.name, + record: record.value as Json, + }); + } } if (evt.event === "delete") { await supabase @@ -95,16 +113,27 @@ async function main() { .eq("uri", evt.uri.toString()); } } - if (evt.collection === ids.PubLeafletPublication) { + if (evt.collection === ids.PubLeafletGraphSubscription) { if (evt.event === "create" || evt.event === "update") { let record = PubLeafletGraphSubscription.validateRecord(evt.record); if (!record.success) return; - await supabase.from("publication_subscriptions").upsert({ - uri: evt.uri.toString(), - identity: evt.did, - publication: record.value.publication, - record: record.value as Json, - }); + let { error } = await supabase + .from("publication_subscriptions") + .upsert({ + uri: evt.uri.toString(), + identity: evt.did, + publication: record.value.publication, + record: record.value as Json, + }); + if (error && error.code === "23503") { + await createIdentity(db, { atp_did: evt.did }); + await supabase.from("publication_subscriptions").upsert({ + uri: evt.uri.toString(), + identity: evt.did, + publication: record.value.publication, + record: record.value as Json, + }); + } } if (evt.event === "delete") { await supabase @@ -113,6 +142,15 @@ async function main() { .eq("uri", evt.uri.toString()); } } + if (evt.collection === ids.AppBskyActorProfile) { + //only listen to updates because we should fetch it for the first time when they subscribe! + if (evt.event === "update") { + await supabaseServerClient + .from("bsky_profiles") + .update({ record: evt.record as Json }) + .eq("did", evt.did); + } + } }, onError: (err) => { console.error(err); @@ -120,10 +158,11 @@ async function main() { }); console.log("starting firehose consumer"); firehose.start(); - const cleanup = () => { + const cleanup = async () => { console.log("shutting down firehose..."); - firehose.destroy(); - runner.destroy(); + await client.end(); + await firehose.destroy(); + await runner.destroy(); process.exit(); }; diff --git a/drizzle/relations.ts b/drizzle/relations.ts index c40b821d..c38ffb3d 100644 --- a/drizzle/relations.ts +++ b/drizzle/relations.ts @@ -1,5 +1,26 @@ import { relations } from "drizzle-orm/relations"; -import { entities, facts, entity_sets, permission_tokens, identities, email_subscriptions_to_entity, email_auth_tokens, custom_domains, phone_rsvps_to_entity, custom_domain_routes, poll_votes_on_entity, subscribers_to_publications, publications, permission_token_on_homepage, documents, documents_in_publications, publication_domains, publication_subscriptions, leaflets_in_publications, permission_token_rights } from "./schema"; +import { identities, bsky_profiles, entities, facts, entity_sets, permission_tokens, email_subscriptions_to_entity, email_auth_tokens, custom_domains, phone_rsvps_to_entity, custom_domain_routes, poll_votes_on_entity, subscribers_to_publications, publications, permission_token_on_homepage, documents, documents_in_publications, publication_domains, publication_subscriptions, leaflets_in_publications, permission_token_rights } from "./schema"; + +export const bsky_profilesRelations = relations(bsky_profiles, ({one}) => ({ + identity: one(identities, { + fields: [bsky_profiles.did], + references: [identities.atp_did] + }), +})); + +export const identitiesRelations = relations(identities, ({one, many}) => ({ + bsky_profiles: many(bsky_profiles), + permission_token: one(permission_tokens, { + fields: [identities.home_page], + references: [permission_tokens.id] + }), + email_auth_tokens: many(email_auth_tokens), + custom_domains: many(custom_domains), + subscribers_to_publications: many(subscribers_to_publications), + permission_token_on_homepages: many(permission_token_on_homepage), + publication_domains: many(publication_domains), + publication_subscriptions: many(publication_subscriptions), +})); export const factsRelations = relations(facts, ({one}) => ({ entity: one(entities, { @@ -48,18 +69,6 @@ export const permission_tokensRelations = relations(permission_tokens, ({one, ma permission_token_rights: many(permission_token_rights), })); -export const identitiesRelations = relations(identities, ({one, many}) => ({ - permission_token: one(permission_tokens, { - fields: [identities.home_page], - references: [permission_tokens.id] - }), - email_auth_tokens: many(email_auth_tokens), - custom_domains: many(custom_domains), - subscribers_to_publications: many(subscribers_to_publications), - permission_token_on_homepages: many(permission_token_on_homepage), - publication_domains: many(publication_domains), -})); - export const email_subscriptions_to_entityRelations = relations(email_subscriptions_to_entity, ({one}) => ({ entity: one(entities, { fields: [email_subscriptions_to_entity.entity], @@ -186,6 +195,10 @@ export const publication_domainsRelations = relations(publication_domains, ({one })); export const publication_subscriptionsRelations = relations(publication_subscriptions, ({one}) => ({ + identity: one(identities, { + fields: [publication_subscriptions.identity], + references: [identities.atp_did] + }), publication: one(publications, { fields: [publication_subscriptions.publication], references: [publications.uri] diff --git a/drizzle/schema.ts b/drizzle/schema.ts index cddbb80c..b4d00f48 100644 --- a/drizzle/schema.ts +++ b/drizzle/schema.ts @@ -1,4 +1,4 @@ -import { pgTable, pgEnum, text, jsonb, timestamp, foreignKey, uuid, bigint, boolean, unique, uniqueIndex, smallint, primaryKey } from "drizzle-orm/pg-core" +import { pgTable, pgEnum, text, jsonb, foreignKey, timestamp, uuid, bigint, boolean, unique, uniqueIndex, smallint, primaryKey } from "drizzle-orm/pg-core" import { sql } from "drizzle-orm" export const aal_level = pgEnum("aal_level", ['aal1', 'aal2', 'aal3']) @@ -24,6 +24,12 @@ export const oauth_session_store = pgTable("oauth_session_store", { session: jsonb("session").notNull(), }); +export const bsky_profiles = pgTable("bsky_profiles", { + did: text("did").primaryKey().notNull().references(() => identities.atp_did, { onDelete: "cascade" } ), + record: jsonb("record").notNull(), + indexed_at: timestamp("indexed_at", { withTimezone: true, mode: 'string' }).defaultNow().notNull(), +}); + export const publications = pgTable("publications", { uri: text("uri").primaryKey().notNull(), indexed_at: timestamp("indexed_at", { withTimezone: true, mode: 'string' }).defaultNow().notNull(), @@ -207,7 +213,7 @@ export const publication_domains = pgTable("publication_domains", { export const publication_subscriptions = pgTable("publication_subscriptions", { publication: text("publication").notNull().references(() => publications.uri, { onDelete: "cascade" } ), - identity: text("identity").notNull(), + identity: text("identity").notNull().references(() => identities.atp_did, { onDelete: "cascade" } ), created_at: timestamp("created_at", { withTimezone: true, mode: 'string' }).defaultNow().notNull(), record: jsonb("record").notNull(), uri: text("uri").notNull(), @@ -220,9 +226,9 @@ export const publication_subscriptions = pgTable("publication_subscriptions", { }); export const leaflets_in_publications = pgTable("leaflets_in_publications", { - publication: text("publication").notNull().references(() => publications.uri), + publication: text("publication").notNull().references(() => publications.uri, { onDelete: "cascade" } ), doc: text("doc").default('').references(() => documents.uri, { onDelete: "set null" } ), - leaflet: uuid("leaflet").notNull().references(() => permission_tokens.id), + leaflet: uuid("leaflet").notNull().references(() => permission_tokens.id, { onDelete: "cascade" } ), description: text("description").default('').notNull(), title: text("title").default('').notNull(), }, diff --git a/lexicons/api/index.ts b/lexicons/api/index.ts index d365af15..3b1bcc7f 100644 --- a/lexicons/api/index.ts +++ b/lexicons/api/index.ts @@ -27,6 +27,7 @@ import * as ComAtprotoRepoListRecords from './types/com/atproto/repo/listRecords import * as ComAtprotoRepoPutRecord from './types/com/atproto/repo/putRecord' import * as ComAtprotoRepoStrongRef from './types/com/atproto/repo/strongRef' import * as ComAtprotoRepoUploadBlob from './types/com/atproto/repo/uploadBlob' +import * as AppBskyActorProfile from './types/app/bsky/actor/profile' export * as PubLeafletDocument from './types/pub/leaflet/document' export * as PubLeafletPublication from './types/pub/leaflet/publication' @@ -50,6 +51,7 @@ export * as ComAtprotoRepoListRecords from './types/com/atproto/repo/listRecords export * as ComAtprotoRepoPutRecord from './types/com/atproto/repo/putRecord' export * as ComAtprotoRepoStrongRef from './types/com/atproto/repo/strongRef' export * as ComAtprotoRepoUploadBlob from './types/com/atproto/repo/uploadBlob' +export * as AppBskyActorProfile from './types/app/bsky/actor/profile' export const PUB_LEAFLET_PAGES = { LinearDocumentTextAlignLeft: 'pub.leaflet.pages.linearDocument#textAlignLeft', @@ -62,11 +64,13 @@ export const PUB_LEAFLET_PAGES = { export class AtpBaseClient extends XrpcClient { pub: PubNS com: ComNS + app: AppNS constructor(options: FetchHandler | FetchHandlerOptions) { super(options, schemas) this.pub = new PubNS(this) this.com = new ComNS(this) + this.app = new AppNS(this) } /** @deprecated use `this` instead */ @@ -472,3 +476,99 @@ export class ComAtprotoRepoNS { ) } } + +export class AppNS { + _client: XrpcClient + bsky: AppBskyNS + + constructor(client: XrpcClient) { + this._client = client + this.bsky = new AppBskyNS(client) + } +} + +export class AppBskyNS { + _client: XrpcClient + actor: AppBskyActorNS + + constructor(client: XrpcClient) { + this._client = client + this.actor = new AppBskyActorNS(client) + } +} + +export class AppBskyActorNS { + _client: XrpcClient + profile: ProfileRecord + + constructor(client: XrpcClient) { + this._client = client + this.profile = new ProfileRecord(client) + } +} + +export class ProfileRecord { + _client: XrpcClient + + constructor(client: XrpcClient) { + this._client = client + } + + async list( + params: OmitKey, + ): Promise<{ + cursor?: string + records: { uri: string; value: AppBskyActorProfile.Record }[] + }> { + const res = await this._client.call('com.atproto.repo.listRecords', { + collection: 'app.bsky.actor.profile', + ...params, + }) + return res.data + } + + async get( + params: OmitKey, + ): Promise<{ uri: string; cid: string; value: AppBskyActorProfile.Record }> { + const res = await this._client.call('com.atproto.repo.getRecord', { + collection: 'app.bsky.actor.profile', + ...params, + }) + return res.data + } + + async create( + params: OmitKey< + ComAtprotoRepoCreateRecord.InputSchema, + 'collection' | 'record' + >, + record: Un$Typed, + headers?: Record, + ): Promise<{ uri: string; cid: string }> { + const collection = 'app.bsky.actor.profile' + const res = await this._client.call( + 'com.atproto.repo.createRecord', + undefined, + { + collection, + rkey: 'self', + ...params, + record: { ...record, $type: collection }, + }, + { encoding: 'application/json', headers }, + ) + return res.data + } + + async delete( + params: OmitKey, + headers?: Record, + ): Promise { + await this._client.call( + 'com.atproto.repo.deleteRecord', + undefined, + { collection: 'app.bsky.actor.profile', ...params }, + { headers }, + ) + } +} diff --git a/lexicons/api/lexicons.ts b/lexicons/api/lexicons.ts index 813b380f..5099d8b2 100644 --- a/lexicons/api/lexicons.ts +++ b/lexicons/api/lexicons.ts @@ -1328,6 +1328,65 @@ export const schemaDict = { }, }, }, + AppBskyActorProfile: { + lexicon: 1, + id: 'app.bsky.actor.profile', + defs: { + main: { + type: 'record', + description: 'A declaration of a Bluesky account profile.', + key: 'literal:self', + record: { + type: 'object', + properties: { + displayName: { + type: 'string', + maxGraphemes: 64, + maxLength: 640, + }, + description: { + type: 'string', + description: 'Free-form profile description text.', + maxGraphemes: 256, + maxLength: 2560, + }, + avatar: { + type: 'blob', + description: + "Small image to be displayed next to posts from account. AKA, 'profile picture'", + accept: ['image/png', 'image/jpeg'], + maxSize: 1000000, + }, + banner: { + type: 'blob', + description: + 'Larger horizontal image to display behind profile view.', + accept: ['image/png', 'image/jpeg'], + maxSize: 1000000, + }, + labels: { + type: 'union', + description: + 'Self-label values, specific to the Bluesky application, on the overall account.', + refs: ['lex:com.atproto.label.defs#selfLabels'], + }, + joinedViaStarterPack: { + type: 'ref', + ref: 'lex:com.atproto.repo.strongRef', + }, + pinnedPost: { + type: 'ref', + ref: 'lex:com.atproto.repo.strongRef', + }, + createdAt: { + type: 'string', + format: 'datetime', + }, + }, + }, + }, + }, + }, } as const satisfies Record export const schemas = Object.values(schemaDict) satisfies LexiconDoc[] @@ -1384,4 +1443,5 @@ export const ids = { ComAtprotoRepoPutRecord: 'com.atproto.repo.putRecord', ComAtprotoRepoStrongRef: 'com.atproto.repo.strongRef', ComAtprotoRepoUploadBlob: 'com.atproto.repo.uploadBlob', + AppBskyActorProfile: 'app.bsky.actor.profile', } as const diff --git a/lexicons/api/types/app/bsky/actor/profile.ts b/lexicons/api/types/app/bsky/actor/profile.ts new file mode 100644 index 00000000..c079f794 --- /dev/null +++ b/lexicons/api/types/app/bsky/actor/profile.ts @@ -0,0 +1,39 @@ +/** + * GENERATED CODE - DO NOT MODIFY + */ +import { ValidationResult, BlobRef } from '@atproto/lexicon' +import { CID } from 'multiformats/cid' +import { validate as _validate } from '../../../../lexicons' +import { $Typed, is$typed as _is$typed, OmitKey } from '../../../../util' +import type * as ComAtprotoLabelDefs from '../../../com/atproto/label/defs' +import type * as ComAtprotoRepoStrongRef from '../../../com/atproto/repo/strongRef' + +const is$typed = _is$typed, + validate = _validate +const id = 'app.bsky.actor.profile' + +export interface Record { + $type: 'app.bsky.actor.profile' + displayName?: string + /** Free-form profile description text. */ + description?: string + /** Small image to be displayed next to posts from account. AKA, 'profile picture' */ + avatar?: BlobRef + /** Larger horizontal image to display behind profile view. */ + banner?: BlobRef + labels?: $Typed | { $type: string } + joinedViaStarterPack?: ComAtprotoRepoStrongRef.Main + pinnedPost?: ComAtprotoRepoStrongRef.Main + createdAt?: string + [k: string]: unknown +} + +const hashRecord = 'main' + +export function isRecord(v: V) { + return is$typed(v, id, hashRecord) +} + +export function validateRecord(v: V) { + return validate(v, id, hashRecord, true) +} diff --git a/lexicons/app/bsky/actor/profile.json b/lexicons/app/bsky/actor/profile.json new file mode 100644 index 00000000..911d7a05 --- /dev/null +++ b/lexicons/app/bsky/actor/profile.json @@ -0,0 +1,53 @@ +{ + "lexicon": 1, + "id": "app.bsky.actor.profile", + "defs": { + "main": { + "type": "record", + "description": "A declaration of a Bluesky account profile.", + "key": "literal:self", + "record": { + "type": "object", + "properties": { + "displayName": { + "type": "string", + "maxGraphemes": 64, + "maxLength": 640 + }, + "description": { + "type": "string", + "description": "Free-form profile description text.", + "maxGraphemes": 256, + "maxLength": 2560 + }, + "avatar": { + "type": "blob", + "description": "Small image to be displayed next to posts from account. AKA, 'profile picture'", + "accept": ["image/png", "image/jpeg"], + "maxSize": 1000000 + }, + "banner": { + "type": "blob", + "description": "Larger horizontal image to display behind profile view.", + "accept": ["image/png", "image/jpeg"], + "maxSize": 1000000 + }, + "labels": { + "type": "union", + "description": "Self-label values, specific to the Bluesky application, on the overall account.", + "refs": ["com.atproto.label.defs#selfLabels"] + }, + "joinedViaStarterPack": { + "type": "ref", + "ref": "com.atproto.repo.strongRef" + }, + "pinnedPost": { + "type": "ref", + "ref": "com.atproto.repo.strongRef" + }, + "createdAt": { "type": "string", "format": "datetime" } + } + } + } + } +} diff --git a/supabase/database.types.ts b/supabase/database.types.ts index 51ed7e81..a4b87174 100644 --- a/supabase/database.types.ts +++ b/supabase/database.types.ts @@ -34,6 +34,32 @@ export type Database = { } public: { Tables: { + bsky_profiles: { + Row: { + did: string + indexed_at: string + record: Json + } + Insert: { + did: string + indexed_at?: string + record: Json + } + Update: { + did?: string + indexed_at?: string + record?: Json + } + Relationships: [ + { + foreignKeyName: "bsky_profiles_did_fkey" + columns: ["did"] + isOneToOne: true + referencedRelation: "identities" + referencedColumns: ["atp_did"] + }, + ] + } custom_domain_routes: { Row: { created_at: string @@ -707,6 +733,13 @@ export type Database = { uri?: string } Relationships: [ + { + foreignKeyName: "publication_subscriptions_identity_fkey" + columns: ["identity"] + isOneToOne: false + referencedRelation: "identities" + referencedColumns: ["atp_did"] + }, { foreignKeyName: "publication_subscriptions_publication_fkey" columns: ["publication"] diff --git a/supabase/migrations/20250610232213_add_bsky_profiles_and_foreign_key_to_subscriptions.sql b/supabase/migrations/20250610232213_add_bsky_profiles_and_foreign_key_to_subscriptions.sql new file mode 100644 index 00000000..18722585 --- /dev/null +++ b/supabase/migrations/20250610232213_add_bsky_profiles_and_foreign_key_to_subscriptions.sql @@ -0,0 +1,61 @@ +create table "public"."bsky_profiles" ( + "did" text not null, + "record" jsonb not null, + "indexed_at" timestamp with time zone not null default now() +); + +alter table "public"."bsky_profiles" enable row level security; + +CREATE UNIQUE INDEX bsky_profiles_pkey ON public.bsky_profiles USING btree (did); + +alter table "public"."bsky_profiles" add constraint "bsky_profiles_pkey" PRIMARY KEY using index "bsky_profiles_pkey"; + +alter table "public"."bsky_profiles" add constraint "bsky_profiles_did_fkey" FOREIGN KEY (did) REFERENCES identities(atp_did) ON DELETE CASCADE not valid; + +alter table "public"."bsky_profiles" validate constraint "bsky_profiles_did_fkey"; + +alter table "public"."publication_subscriptions" add constraint "publication_subscriptions_identity_fkey" FOREIGN KEY (identity) REFERENCES identities(atp_did) ON DELETE CASCADE not valid; + +alter table "public"."publication_subscriptions" validate constraint "publication_subscriptions_identity_fkey"; + +grant delete on table "public"."bsky_profiles" to "anon"; + +grant insert on table "public"."bsky_profiles" to "anon"; + +grant references on table "public"."bsky_profiles" to "anon"; + +grant select on table "public"."bsky_profiles" to "anon"; + +grant trigger on table "public"."bsky_profiles" to "anon"; + +grant truncate on table "public"."bsky_profiles" to "anon"; + +grant update on table "public"."bsky_profiles" to "anon"; + +grant delete on table "public"."bsky_profiles" to "authenticated"; + +grant insert on table "public"."bsky_profiles" to "authenticated"; + +grant references on table "public"."bsky_profiles" to "authenticated"; + +grant select on table "public"."bsky_profiles" to "authenticated"; + +grant trigger on table "public"."bsky_profiles" to "authenticated"; + +grant truncate on table "public"."bsky_profiles" to "authenticated"; + +grant update on table "public"."bsky_profiles" to "authenticated"; + +grant delete on table "public"."bsky_profiles" to "service_role"; + +grant insert on table "public"."bsky_profiles" to "service_role"; + +grant references on table "public"."bsky_profiles" to "service_role"; + +grant select on table "public"."bsky_profiles" to "service_role"; + +grant trigger on table "public"."bsky_profiles" to "service_role"; + +grant truncate on table "public"."bsky_profiles" to "service_role"; + +grant update on table "public"."bsky_profiles" to "service_role";