graham.systems
Encrypted, ephemeral, private memos on atproto
refactor(consumer): extract class from mod.ts
+219
packages/consumer/client.ts
+219
packages/consumer/client.ts
···
1
+
import {
2
+
produceRequirements,
3
+
type XRPCProcedures,
4
+
type XRPCQueries,
5
+
} from "@cistern/shared";
6
+
import { decryptText, generateKeys } from "@cistern/crypto";
7
+
import { generateRandomName } from "@puregarlic/randimal";
8
+
import { is, parse, type RecordKey } from "@atcute/lexicons";
9
+
import { JetstreamSubscription } from "@atcute/jetstream";
10
+
import type { Did } from "@atcute/lexicons/syntax";
11
+
import type { Client } from "@atcute/client";
12
+
import { AppCisternMemo, type AppCisternPubkey } from "@cistern/lexicon";
13
+
import type {
14
+
ConsumerOptions,
15
+
ConsumerParams,
16
+
DecryptedMemo,
17
+
LocalKeyPair,
18
+
} from "./types.ts";
19
+
20
+
/**
21
+
* Client for generating keys and decoding Cistern memos.
22
+
*/
23
+
export class Consumer {
24
+
/** DID of the user this consumer acts on behalf of */
25
+
did: Did;
26
+
27
+
/** `@atcute/client` instance with credential manager */
28
+
rpc: Client<XRPCQueries, XRPCProcedures>;
29
+
30
+
/** Private key used for decrypting and the AT URI of its associated public key */
31
+
keypair?: LocalKeyPair;
32
+
33
+
constructor(params: ConsumerParams) {
34
+
this.did = params.miniDoc.did;
35
+
this.keypair = params.options.keypair
36
+
? {
37
+
privateKey: Uint8Array.fromBase64(params.options.keypair.privateKey),
38
+
publicKey: params.options.keypair.publicKey,
39
+
}
40
+
: undefined;
41
+
this.rpc = params.rpc;
42
+
}
43
+
44
+
/**
45
+
* Generates a key pair, uploading the public key to PDS and returning the pair.
46
+
*/
47
+
async generateKeyPair(): Promise<LocalKeyPair> {
48
+
if (this.keypair) {
49
+
throw new Error("client already has a key pair");
50
+
}
51
+
52
+
const keys = generateKeys();
53
+
const name = await generateRandomName();
54
+
55
+
const record: AppCisternPubkey.Main = {
56
+
$type: "app.cistern.pubkey",
57
+
name,
58
+
algorithm: "x_wing",
59
+
content: { $bytes: keys.publicKey.toBase64() },
60
+
createdAt: new Date().toISOString(),
61
+
};
62
+
const res = await this.rpc.post("com.atproto.repo.createRecord", {
63
+
input: {
64
+
collection: "app.cistern.pubkey",
65
+
repo: this.did,
66
+
record,
67
+
},
68
+
});
69
+
70
+
if (!res.ok) {
71
+
throw new Error(
72
+
`failed to save public key: ${res.status} ${res.data.error}`,
73
+
);
74
+
}
75
+
76
+
const keypair = {
77
+
privateKey: keys.secretKey,
78
+
publicKey: res.data.uri,
79
+
};
80
+
81
+
this.keypair = keypair;
82
+
83
+
return keypair;
84
+
}
85
+
86
+
/**
87
+
* Asynchronously iterate through memos in the user's PDS
88
+
*/
89
+
async *listMemos(): AsyncGenerator<
90
+
DecryptedMemo,
91
+
void,
92
+
undefined
93
+
> {
94
+
if (!this.keypair) {
95
+
throw new Error("no key pair set; generate a key before listing memos");
96
+
}
97
+
98
+
let cursor: string | undefined;
99
+
100
+
while (true) {
101
+
const res = await this.rpc.get("com.atproto.repo.listRecords", {
102
+
params: {
103
+
collection: "app.cistern.memo",
104
+
repo: this.did,
105
+
cursor,
106
+
},
107
+
});
108
+
109
+
if (!res.ok) {
110
+
throw new Error(
111
+
`failed to list memos: ${res.status} ${res.data.error}`,
112
+
);
113
+
}
114
+
115
+
cursor = res.data.cursor;
116
+
117
+
for (const record of res.data.records) {
118
+
const memo = parse(AppCisternMemo.mainSchema, record.value);
119
+
120
+
if (memo.pubkey !== this.keypair.publicKey) continue;
121
+
122
+
const decrypted = decryptText(this.keypair.privateKey, {
123
+
nonce: memo.nonce.$bytes,
124
+
cipherText: memo.ciphertext.$bytes,
125
+
content: memo.payload.$bytes,
126
+
hash: memo.contentHash.$bytes,
127
+
length: memo.contentLength,
128
+
});
129
+
130
+
yield {
131
+
tid: memo.tid,
132
+
text: decrypted,
133
+
};
134
+
}
135
+
136
+
if (!cursor) return;
137
+
}
138
+
}
139
+
140
+
/**
141
+
* Subscribes to the Jetstreams for the user's memos. Pass `"stop"` into `subscription.next(...)` to cancel
142
+
* @todo Allow specifying Jetstream endpoint
143
+
*/
144
+
async *subscribeToMemos(): AsyncGenerator<
145
+
DecryptedMemo,
146
+
void,
147
+
"stop" | undefined
148
+
> {
149
+
if (!this.keypair) {
150
+
throw new Error("no key pair set; generate a key before subscribing");
151
+
}
152
+
153
+
const subscription = new JetstreamSubscription({
154
+
url: "wss://jetstream2.us-east.bsky.network",
155
+
wantedCollections: ["app.cistern.memo"],
156
+
wantedDids: [this.did],
157
+
});
158
+
159
+
for await (const event of subscription) {
160
+
if (event.kind === "commit" && event.commit.operation === "create") {
161
+
const record = event.commit.record;
162
+
163
+
if (!is(AppCisternMemo.mainSchema, record)) {
164
+
continue;
165
+
}
166
+
167
+
if (record.pubkey !== this.keypair.publicKey) {
168
+
continue;
169
+
}
170
+
171
+
const decrypted = decryptText(this.keypair.privateKey, {
172
+
nonce: record.nonce.$bytes,
173
+
cipherText: record.ciphertext.$bytes,
174
+
content: record.payload.$bytes,
175
+
hash: record.contentHash.$bytes,
176
+
length: record.contentLength,
177
+
});
178
+
179
+
const command = yield { tid: record.tid, text: decrypted };
180
+
181
+
if (command === "stop") return;
182
+
}
183
+
}
184
+
}
185
+
186
+
/**
187
+
* Deletes a memo from the user's PDS by record key.
188
+
*/
189
+
async deleteMemo(key: RecordKey) {
190
+
const res = await this.rpc.post("com.atproto.repo.deleteRecord", {
191
+
input: {
192
+
collection: "app.cistern.memo",
193
+
repo: this.did,
194
+
rkey: key,
195
+
},
196
+
});
197
+
198
+
if (!res.ok) {
199
+
throw new Error(
200
+
`failed to delete memo ${key}: ${res.status} ${res.data.error}`,
201
+
);
202
+
}
203
+
}
204
+
}
205
+
206
+
/**
207
+
* Creates a `Consumer` instance with all necessary requirements. This is the recommended way to construct a `Consumer`.
208
+
*
209
+
* @description Resolves the user's DID using Slingshot, instantiates an `@atcute/client` instance, creates an initial session, and then returns a new Consumer.
210
+
* @param {ConsumerOptions} options - Information for constructing the underlying XRPC client
211
+
* @returns {Promise<Consumer>} A Cistern consumer client with an authorized session
212
+
*/
213
+
export async function createConsumer(
214
+
options: ConsumerOptions,
215
+
): Promise<Consumer> {
216
+
const reqs = await produceRequirements(options);
217
+
218
+
return new Consumer(reqs);
219
+
}
+2
-219
packages/consumer/mod.ts
+2
-219
packages/consumer/mod.ts
···
1
-
import {
2
-
produceRequirements,
3
-
type XRPCProcedures,
4
-
type XRPCQueries,
5
-
} from "@cistern/shared";
6
-
import { decryptText, generateKeys } from "@cistern/crypto";
7
-
import { generateRandomName } from "@puregarlic/randimal";
8
-
import { is, parse, type RecordKey } from "@atcute/lexicons";
9
-
import { JetstreamSubscription } from "@atcute/jetstream";
10
-
import type { Did } from "@atcute/lexicons/syntax";
11
-
import type { Client, CredentialManager } from "@atcute/client";
12
-
import { AppCisternMemo, type AppCisternPubkey } from "@cistern/lexicon";
13
-
import type {
14
-
ConsumerOptions,
15
-
ConsumerParams,
16
-
DecryptedMemo,
17
-
LocalKeyPair,
18
-
} from "./types.ts";
19
-
20
-
/**
21
-
* Creates a `Consumer` instance with all necessary requirements. This is the recommended way to construct a `Consumer`.
22
-
*
23
-
* @description Resolves the user's DID using Slingshot, instantiates an `@atcute/client` instance, creates an initial session, and then returns a new Consumer.
24
-
* @param {ConsumerOptions} options - Information for constructing the underlying XRPC client
25
-
* @returns {Promise<Consumer>} A Cistern consumer client with an authorized session
26
-
*/
27
-
export async function createConsumer(
28
-
options: ConsumerOptions,
29
-
): Promise<Consumer> {
30
-
const reqs = await produceRequirements(options);
31
-
32
-
return new Consumer(reqs);
33
-
}
34
-
35
-
/**
36
-
* Client for generating keys and decoding Cistern memos.
37
-
*/
38
-
export class Consumer {
39
-
/** DID of the user this consumer acts on behalf of */
40
-
did: Did;
41
-
42
-
/** `@atcute/client` instance with credential manager */
43
-
rpc: Client<XRPCQueries, XRPCProcedures>;
44
-
45
-
/** Private key used for decrypting and the AT URI of its associated public key */
46
-
keypair?: LocalKeyPair;
47
-
48
-
constructor(params: ConsumerParams) {
49
-
this.did = params.miniDoc.did;
50
-
this.keypair = params.options.keypair
51
-
? {
52
-
privateKey: Uint8Array.fromBase64(params.options.keypair.privateKey),
53
-
publicKey: params.options.keypair.publicKey,
54
-
}
55
-
: undefined;
56
-
this.rpc = params.rpc;
57
-
}
58
-
59
-
/**
60
-
* Generates a key pair, uploading the public key to PDS and returning the pair.
61
-
*/
62
-
async generateKeyPair(): Promise<LocalKeyPair> {
63
-
if (this.keypair) {
64
-
throw new Error("client already has a key pair");
65
-
}
66
-
67
-
const keys = generateKeys();
68
-
const name = await generateRandomName();
69
-
70
-
const record: AppCisternPubkey.Main = {
71
-
$type: "app.cistern.pubkey",
72
-
name,
73
-
algorithm: "x_wing",
74
-
content: { $bytes: keys.publicKey.toBase64() },
75
-
createdAt: new Date().toISOString(),
76
-
};
77
-
const res = await this.rpc.post("com.atproto.repo.createRecord", {
78
-
input: {
79
-
collection: "app.cistern.pubkey",
80
-
repo: this.did,
81
-
record,
82
-
},
83
-
});
84
-
85
-
if (!res.ok) {
86
-
throw new Error(
87
-
`failed to save public key: ${res.status} ${res.data.error}`,
88
-
);
89
-
}
90
-
91
-
const keypair = {
92
-
privateKey: keys.secretKey,
93
-
publicKey: res.data.uri,
94
-
};
95
-
96
-
this.keypair = keypair;
97
-
98
-
return keypair;
99
-
}
100
-
101
-
/**
102
-
* Asynchronously iterate through memos in the user's PDS
103
-
*/
104
-
async *listMemos(): AsyncGenerator<
105
-
DecryptedMemo,
106
-
void,
107
-
undefined
108
-
> {
109
-
if (!this.keypair) {
110
-
throw new Error("no key pair set; generate a key before listing memos");
111
-
}
112
-
113
-
let cursor: string | undefined;
114
-
115
-
while (true) {
116
-
const res = await this.rpc.get("com.atproto.repo.listRecords", {
117
-
params: {
118
-
collection: "app.cistern.memo",
119
-
repo: this.did,
120
-
cursor,
121
-
},
122
-
});
123
-
124
-
if (!res.ok) {
125
-
throw new Error(
126
-
`failed to list memos: ${res.status} ${res.data.error}`,
127
-
);
128
-
}
129
-
130
-
cursor = res.data.cursor;
131
-
132
-
for (const record of res.data.records) {
133
-
const memo = parse(AppCisternMemo.mainSchema, record.value);
134
-
135
-
if (memo.pubkey !== this.keypair.publicKey) continue;
136
-
137
-
const decrypted = decryptText(this.keypair.privateKey, {
138
-
nonce: memo.nonce.$bytes,
139
-
cipherText: memo.ciphertext.$bytes,
140
-
content: memo.payload.$bytes,
141
-
hash: memo.contentHash.$bytes,
142
-
length: memo.contentLength,
143
-
});
144
-
145
-
yield {
146
-
tid: memo.tid,
147
-
text: decrypted,
148
-
};
149
-
}
150
-
151
-
if (!cursor) return;
152
-
}
153
-
}
154
-
155
-
/**
156
-
* Subscribes to the Jetstreams for the user's memos. Pass `"stop"` into `subscription.next(...)` to cancel
157
-
* @todo Allow specifying Jetstream endpoint
158
-
*/
159
-
async *subscribeToMemos(): AsyncGenerator<
160
-
DecryptedMemo,
161
-
void,
162
-
"stop" | undefined
163
-
> {
164
-
if (!this.keypair) {
165
-
throw new Error("no key pair set; generate a key before subscribing");
166
-
}
167
-
168
-
const subscription = new JetstreamSubscription({
169
-
url: "wss://jetstream2.us-east.bsky.network",
170
-
wantedCollections: ["app.cistern.memo"],
171
-
wantedDids: [this.did],
172
-
});
173
-
174
-
for await (const event of subscription) {
175
-
if (event.kind === "commit" && event.commit.operation === "create") {
176
-
const record = event.commit.record;
177
-
178
-
if (!is(AppCisternMemo.mainSchema, record)) {
179
-
continue;
180
-
}
181
-
182
-
if (record.pubkey !== this.keypair.publicKey) {
183
-
continue;
184
-
}
185
-
186
-
const decrypted = decryptText(this.keypair.privateKey, {
187
-
nonce: record.nonce.$bytes,
188
-
cipherText: record.ciphertext.$bytes,
189
-
content: record.payload.$bytes,
190
-
hash: record.contentHash.$bytes,
191
-
length: record.contentLength,
192
-
});
193
-
194
-
const command = yield { tid: record.tid, text: decrypted };
195
-
196
-
if (command === "stop") return;
197
-
}
198
-
}
199
-
}
200
-
201
-
/**
202
-
* Deletes a memo from the user's PDS by record key.
203
-
*/
204
-
async deleteMemo(key: RecordKey) {
205
-
const res = await this.rpc.post("com.atproto.repo.deleteRecord", {
206
-
input: {
207
-
collection: "app.cistern.memo",
208
-
repo: this.did,
209
-
rkey: key,
210
-
},
211
-
});
212
-
213
-
if (!res.ok) {
214
-
throw new Error(
215
-
`failed to delete memo ${key}: ${res.status} ${res.data.error}`,
216
-
);
217
-
}
218
-
}
219
-
}
1
+
export * from "./client.ts";
2
+
export * from "./types.ts";