From eb7906e709ab05eb0ce48fed86773f79e679ad50 Mon Sep 17 00:00:00 2001 From: Wesley Finck Date: Fri, 27 Feb 2026 10:44:58 -0800 Subject: [PATCH] connection lexicon and publisher --- .../infrastructure/lexicon/lexicons.ts | 48 ++++++ .../types/network/cosmik/connection.ts | 42 +++++ .../infrastructure/lexicons/connection.json | 45 ++++++ .../mappers/ConnectionMapper.ts | 22 +++ .../publishers/ATProtoConnectionPublisher.ts | 150 ++++++++++++++++++ .../commands/CreateConnectionUseCase.ts | 13 +- .../commands/DeleteConnectionUseCase.ts | 5 +- .../tests/utils/FakeConnectionPublisher.ts | 120 ++++++++++++++ src/shared/constants/atproto.ts | 1 + .../config/EnvironmentConfigService.ts | 5 + 10 files changed, 439 insertions(+), 12 deletions(-) create mode 100644 src/modules/atproto/infrastructure/lexicon/types/network/cosmik/connection.ts create mode 100644 src/modules/atproto/infrastructure/lexicons/connection.json create mode 100644 src/modules/atproto/infrastructure/mappers/ConnectionMapper.ts create mode 100644 src/modules/atproto/infrastructure/publishers/ATProtoConnectionPublisher.ts create mode 100644 src/modules/cards/tests/utils/FakeConnectionPublisher.ts diff --git a/src/modules/atproto/infrastructure/lexicon/lexicons.ts b/src/modules/atproto/infrastructure/lexicon/lexicons.ts index b2b65c60..bc3262a8 100644 --- a/src/modules/atproto/infrastructure/lexicon/lexicons.ts +++ b/src/modules/atproto/infrastructure/lexicon/lexicons.ts @@ -297,6 +297,53 @@ export const schemaDict = { }, }, }, + NetworkCosmikConnection: { + lexicon: 1, + id: 'network.cosmik.connection', + description: + 'A record representing a connection between two entities (URLs or cards).', + defs: { + main: { + type: 'record', + description: + 'A connection linking a source to a target, with optional type and note.', + key: 'tid', + record: { + type: 'object', + required: ['source', 'target'], + properties: { + source: { + type: 'string', + description: 'Source entity (URL string or AT URI)', + }, + target: { + type: 'string', + description: 'Target entity (URL string or AT URI)', + }, + connectionType: { + type: 'string', + description: 'Optional type of connection', + }, + note: { + type: 'string', + description: 'Optional note about the connection', + maxLength: 1000, + }, + createdAt: { + type: 'string', + format: 'datetime', + description: 'Timestamp when this connection was created.', + }, + updatedAt: { + type: 'string', + format: 'datetime', + description: 'Timestamp when this connection was last updated.', + }, + }, + }, + }, + }, + }, NetworkCosmikDefs: { lexicon: 1, id: 'network.cosmik.defs', @@ -990,6 +1037,7 @@ export const ids = { NetworkCosmikCollection: 'network.cosmik.collection', NetworkCosmikCollectionLink: 'network.cosmik.collectionLink', NetworkCosmikCollectionLinkRemoval: 'network.cosmik.collectionLinkRemoval', + NetworkCosmikConnection: 'network.cosmik.connection', NetworkCosmikDefs: 'network.cosmik.defs', NetworkCosmikFollow: 'network.cosmik.follow', ComAtprotoRepoStrongRef: 'com.atproto.repo.strongRef', diff --git a/src/modules/atproto/infrastructure/lexicon/types/network/cosmik/connection.ts b/src/modules/atproto/infrastructure/lexicon/types/network/cosmik/connection.ts new file mode 100644 index 00000000..3b7f24e4 --- /dev/null +++ b/src/modules/atproto/infrastructure/lexicon/types/network/cosmik/connection.ts @@ -0,0 +1,42 @@ +/** + * GENERATED CODE - DO NOT MODIFY + */ +import { type ValidationResult, BlobRef } from '@atproto/lexicon'; +import { CID } from 'multiformats/cid'; +import { validate as _validate } from '../../../lexicons'; +import { + type $Typed, + is$typed as _is$typed, + type OmitKey, +} from '../../../util'; + +const is$typed = _is$typed, + validate = _validate; +const id = 'network.cosmik.connection'; + +export interface Record { + $type: 'network.cosmik.connection'; + /** Source entity (URL string or AT URI) */ + source: string; + /** Target entity (URL string or AT URI) */ + target: string; + /** Optional type of connection */ + connectionType?: string; + /** Optional note about the connection */ + note?: string; + /** Timestamp when this connection was created. */ + createdAt?: string; + /** Timestamp when this connection was last updated. */ + updatedAt?: 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/src/modules/atproto/infrastructure/lexicons/connection.json b/src/modules/atproto/infrastructure/lexicons/connection.json new file mode 100644 index 00000000..020b2167 --- /dev/null +++ b/src/modules/atproto/infrastructure/lexicons/connection.json @@ -0,0 +1,45 @@ +{ + "lexicon": 1, + "id": "network.cosmik.connection", + "description": "A record representing a connection between two entities (URLs or cards).", + "defs": { + "main": { + "type": "record", + "description": "A connection linking a source to a target, with optional type and note.", + "key": "tid", + "record": { + "type": "object", + "required": ["source", "target"], + "properties": { + "source": { + "type": "string", + "description": "Source entity (URL string or AT URI)" + }, + "target": { + "type": "string", + "description": "Target entity (URL string or AT URI)" + }, + "connectionType": { + "type": "string", + "description": "Optional type of connection" + }, + "note": { + "type": "string", + "description": "Optional note about the connection", + "maxLength": 1000 + }, + "createdAt": { + "type": "string", + "format": "datetime", + "description": "Timestamp when this connection was created." + }, + "updatedAt": { + "type": "string", + "format": "datetime", + "description": "Timestamp when this connection was last updated." + } + } + } + } + } +} diff --git a/src/modules/atproto/infrastructure/mappers/ConnectionMapper.ts b/src/modules/atproto/infrastructure/mappers/ConnectionMapper.ts new file mode 100644 index 00000000..6a7f5a9a --- /dev/null +++ b/src/modules/atproto/infrastructure/mappers/ConnectionMapper.ts @@ -0,0 +1,22 @@ +import { Connection } from 'src/modules/cards/domain/Connection'; +import { Record } from '../lexicon/types/network/cosmik/connection'; +import { EnvironmentConfigService } from 'src/shared/infrastructure/config/EnvironmentConfigService'; + +type ConnectionRecordDTO = Record; + +export class ConnectionMapper { + static readonly connectionType = + new EnvironmentConfigService().getAtProtoCollections().connection; + + static toCreateRecordDTO(connection: Connection): ConnectionRecordDTO { + return { + $type: this.connectionType as any, + source: connection.source.stringValue, + target: connection.target.stringValue, + connectionType: connection.type?.value, + note: connection.note?.value, + createdAt: connection.createdAt.toISOString(), + updatedAt: connection.updatedAt.toISOString(), + }; + } +} diff --git a/src/modules/atproto/infrastructure/publishers/ATProtoConnectionPublisher.ts b/src/modules/atproto/infrastructure/publishers/ATProtoConnectionPublisher.ts new file mode 100644 index 00000000..b32fb20c --- /dev/null +++ b/src/modules/atproto/infrastructure/publishers/ATProtoConnectionPublisher.ts @@ -0,0 +1,150 @@ +import { IConnectionPublisher } from 'src/modules/cards/application/ports/IConnectionPublisher'; +import { Connection } from 'src/modules/cards/domain/Connection'; +import { Result, ok, err } from 'src/shared/core/Result'; +import { UseCaseError } from 'src/shared/core/UseCaseError'; +import { PublishedRecordId } from 'src/modules/cards/domain/value-objects/PublishedRecordId'; +import { ConnectionMapper } from '../mappers/ConnectionMapper'; +import { StrongRef } from '../../domain'; +import { IAgentService } from '../../application/IAgentService'; +import { DID } from '../../domain/DID'; +import { AuthenticationError } from 'src/shared/core/AuthenticationError'; + +export class ATProtoConnectionPublisher implements IConnectionPublisher { + constructor( + private readonly agentService: IAgentService, + private readonly connectionCollection: string, + ) {} + + async publish( + connection: Connection, + ): Promise> { + try { + const curatorDidResult = DID.create(connection.curatorId.value); + + if (curatorDidResult.isErr()) { + return err( + new Error(`Invalid curator DID: ${curatorDidResult.error.message}`), + ); + } + + const curatorDid = curatorDidResult.value; + + // Get an authenticated agent for this curator + const agentResult = + await this.agentService.getAuthenticatedAgent(curatorDid); + + if (agentResult.isErr()) { + // Propagate authentication errors as-is + if (agentResult.error instanceof AuthenticationError) { + return err(agentResult.error); + } + return err( + new Error( + `Authentication error for ATProtoConnectionPublisher: ${agentResult.error.message}`, + ), + ); + } + + const agent = agentResult.value; + + if (!agent) { + return err(new Error('No authenticated session found for curator')); + } + + if (connection.publishedRecordId) { + // Update existing connection record + const connectionRecordDTO = + ConnectionMapper.toCreateRecordDTO(connection); + connectionRecordDTO.$type = this.connectionCollection as any; + + const publishedRecordId = connection.publishedRecordId.getValue(); + const strongRef = new StrongRef(publishedRecordId); + const atUri = strongRef.atUri; + const rkey = atUri.rkey; + + const updateResult = await agent.com.atproto.repo.putRecord({ + repo: curatorDid.value, + collection: this.connectionCollection, + rkey: rkey, + record: connectionRecordDTO, + }); + + return ok( + PublishedRecordId.create({ + uri: updateResult.data.uri, + cid: updateResult.data.cid, + }), + ); + } else { + // Create new connection record + const connectionRecordDTO = + ConnectionMapper.toCreateRecordDTO(connection); + connectionRecordDTO.$type = this.connectionCollection as any; + + const createResult = await agent.com.atproto.repo.createRecord({ + repo: curatorDid.value, + collection: this.connectionCollection, + record: connectionRecordDTO, + }); + + return ok( + PublishedRecordId.create({ + uri: createResult.data.uri, + cid: createResult.data.cid, + }), + ); + } + } catch (error) { + return err( + new Error(error instanceof Error ? error.message : String(error)), + ); + } + } + + async unpublish( + recordId: PublishedRecordId, + ): Promise> { + try { + const publishedRecordId = recordId.getValue(); + const strongRef = new StrongRef(publishedRecordId); + const atUri = strongRef.atUri; + const curatorDid = atUri.did; + const repo = atUri.did.toString(); + const rkey = atUri.rkey; + + // Get an authenticated agent for this curator + const agentResult = + await this.agentService.getAuthenticatedAgent(curatorDid); + + if (agentResult.isErr()) { + // Propagate authentication errors as-is + if (agentResult.error instanceof AuthenticationError) { + return err(agentResult.error); + } + return err( + new Error( + `Authentication error for ATProtoConnectionPublisher: ${agentResult.error.message}`, + ), + ); + } + + const agent = agentResult.value; + + if (!agent) { + return err(new Error('No authenticated session found for curator')); + } + + await agent.com.atproto.repo.deleteRecord({ + repo, + collection: this.connectionCollection, + rkey, + }); + + return ok(undefined); + } catch (error) { + return err( + new Error(error instanceof Error ? error.message : String(error)), + ); + } + } +} diff --git a/src/modules/cards/application/useCases/commands/CreateConnectionUseCase.ts b/src/modules/cards/application/useCases/commands/CreateConnectionUseCase.ts index f544c93a..0049d350 100644 --- a/src/modules/cards/application/useCases/commands/CreateConnectionUseCase.ts +++ b/src/modules/cards/application/useCases/commands/CreateConnectionUseCase.ts @@ -83,9 +83,7 @@ export class CreateConnectionUseCase extends BaseUseCase< ); if (sourceResult.isErr()) { return err( - new ValidationError( - `Invalid source: ${sourceResult.error.message}`, - ), + new ValidationError(`Invalid source: ${sourceResult.error.message}`), ); } const source = sourceResult.value; @@ -97,9 +95,7 @@ export class CreateConnectionUseCase extends BaseUseCase< ); if (targetResult.isErr()) { return err( - new ValidationError( - `Invalid target: ${targetResult.error.message}`, - ), + new ValidationError(`Invalid target: ${targetResult.error.message}`), ); } const target = targetResult.value; @@ -182,9 +178,8 @@ export class CreateConnectionUseCase extends BaseUseCase< } // Publish domain events (ConnectionCreatedEvent) - const publishEventsResult = await this.publishEventsForAggregate( - connection, - ); + const publishEventsResult = + await this.publishEventsForAggregate(connection); if (publishEventsResult.isErr()) { console.error( 'Failed to publish domain events:', diff --git a/src/modules/cards/application/useCases/commands/DeleteConnectionUseCase.ts b/src/modules/cards/application/useCases/commands/DeleteConnectionUseCase.ts index 7eef8429..da185be0 100644 --- a/src/modules/cards/application/useCases/commands/DeleteConnectionUseCase.ts +++ b/src/modules/cards/application/useCases/commands/DeleteConnectionUseCase.ts @@ -126,9 +126,8 @@ export class DeleteConnectionUseCase extends BaseUseCase< } // Publish domain events (ConnectionRemovedEvent) - const publishEventsResult = await this.publishEventsForAggregate( - connection, - ); + const publishEventsResult = + await this.publishEventsForAggregate(connection); if (publishEventsResult.isErr()) { console.error( 'Failed to publish domain events:', diff --git a/src/modules/cards/tests/utils/FakeConnectionPublisher.ts b/src/modules/cards/tests/utils/FakeConnectionPublisher.ts new file mode 100644 index 00000000..834ce8ef --- /dev/null +++ b/src/modules/cards/tests/utils/FakeConnectionPublisher.ts @@ -0,0 +1,120 @@ +import { IConnectionPublisher } from '../../application/ports/IConnectionPublisher'; +import { Connection } from '../../domain/Connection'; +import { PublishedRecordId } from '../../domain/value-objects/PublishedRecordId'; +import { ok, err, Result } from '../../../../shared/core/Result'; +import { UseCaseError } from '../../../../shared/core/UseCaseError'; +import { AppError } from '../../../../shared/core/AppError'; +import { EnvironmentConfigService } from 'src/shared/infrastructure/config/EnvironmentConfigService'; + +export class FakeConnectionPublisher implements IConnectionPublisher { + private publishedConnections: Map = new Map(); + private unpublishedConnections: Array<{ uri: string; cid: string }> = []; + private shouldFail: boolean = false; + private shouldFailUnpublish: boolean = false; + private connectionType = + new EnvironmentConfigService().getAtProtoCollections().connection; + + async publish( + connection: Connection, + ): Promise> { + if (this.shouldFail) { + return err( + AppError.UnexpectedError.create( + new Error('Simulated connection publish failure'), + ), + ); + } + + const connectionId = connection.connectionId.getStringValue(); + + // Use the connection's curator DID directly + const fakeDid = connection.curatorId.value; + + // Simulate publishing the connection record + const fakeConnectionUri = `at://${fakeDid}/${this.connectionType}/${connectionId}`; + const fakeConnectionCid = `fake-connection-cid-${connectionId}`; + + const connectionRecord = PublishedRecordId.create({ + uri: fakeConnectionUri, + cid: fakeConnectionCid, + }); + + // Store the published connection for inspection + this.publishedConnections.set(connectionId, connection); + + console.log( + `[FakeConnectionPublisher] Published connection ${connectionId}`, + ); + + return ok(connectionRecord); + } + + async unpublish( + recordId: PublishedRecordId, + ): Promise> { + if (this.shouldFailUnpublish) { + return err( + AppError.UnexpectedError.create( + new Error('Simulated connection unpublish failure'), + ), + ); + } + + // Find and remove the connection by its published record ID + for (const [ + connectionId, + connection, + ] of this.publishedConnections.entries()) { + if (connection.publishedRecordId?.uri === recordId.uri) { + this.publishedConnections.delete(connectionId); + this.unpublishedConnections.push({ + uri: recordId.uri, + cid: recordId.cid, + }); + console.log( + `[FakeConnectionPublisher] Unpublished connection ${recordId.uri}`, + ); + return ok(undefined); + } + } + + if (this.publishedConnections.size === 0) { + this.unpublishedConnections.push({ + uri: recordId.uri, + cid: recordId.cid, + }); + console.log( + `[FakeConnectionPublisher] Unpublished connection ${recordId.uri} (not found in published connections)`, + ); + return ok(undefined); + } + + console.warn( + `[FakeConnectionPublisher] Connection not found for unpublishing: ${recordId.uri}`, + ); + return ok(undefined); + } + + setShouldFail(shouldFail: boolean): void { + this.shouldFail = shouldFail; + } + + setShouldFailUnpublish(shouldFailUnpublish: boolean): void { + this.shouldFailUnpublish = shouldFailUnpublish; + } + + clear(): void { + this.publishedConnections.clear(); + this.unpublishedConnections = []; + this.shouldFail = false; + this.shouldFailUnpublish = false; + } + + getPublishedConnections(): Connection[] { + return Array.from(this.publishedConnections.values()); + } + + getUnpublishedConnections(): Array<{ uri: string; cid: string }> { + return this.unpublishedConnections; + } +} diff --git a/src/shared/constants/atproto.ts b/src/shared/constants/atproto.ts index cc02d48e..463ddea7 100644 --- a/src/shared/constants/atproto.ts +++ b/src/shared/constants/atproto.ts @@ -16,5 +16,6 @@ export const ATPROTO_NSID = { COLLECTION_LINK: 'network.cosmik.collectionLink', COLLECTION_LINK_REMOVAL: 'network.cosmik.collectionLinkRemoval', FOLLOW: 'network.cosmik.follow', + CONNECTION: 'network.cosmik.connection', }, } as const; diff --git a/src/shared/infrastructure/config/EnvironmentConfigService.ts b/src/shared/infrastructure/config/EnvironmentConfigService.ts index 343b8d78..803202e1 100644 --- a/src/shared/infrastructure/config/EnvironmentConfigService.ts +++ b/src/shared/infrastructure/config/EnvironmentConfigService.ts @@ -37,6 +37,7 @@ export interface EnvironmentConfig { marginCollectionItem: string; collectionLinkRemoval: string; follow: string; + connection: string; }; serviceAccount: { identifier: string; @@ -133,6 +134,10 @@ export class EnvironmentConfigService { environment === Environment.PROD ? ATPROTO_NSID.COSMIK.FOLLOW : `${ATPROTO_NSID.COSMIK.NAMESPACE}.${environment}.follow`, + connection: + environment === Environment.PROD + ? ATPROTO_NSID.COSMIK.CONNECTION + : `${ATPROTO_NSID.COSMIK.NAMESPACE}.${environment}.connection`, // Margin collections - no environment suffix marginBookmark: ATPROTO_NSID.MARGIN.BOOKMARK, marginCollection: ATPROTO_NSID.MARGIN.COLLECTION, -- 2.51.2