The recipes.blue monorepo
recipes.blue
recipes
appview
atproto
feat: implement recipe ingest
+61
-9
+1
-1
apps/api/src/config/env.ts
+1
-1
apps/api/src/config/env.ts
+2
-2
apps/api/src/db/schema.ts
+2
-2
apps/api/src/db/schema.ts
···
24
24
rkey: text('rkey').notNull(),
25
25
title: text('title').notNull(),
26
26
description: text('description'),
27
-
ingredients: text('ingredients', { mode: 'json' }).$type<Ingredient[]>().notNull(),
28
-
steps: text('steps', { mode: 'json' }).$type<Step[]>().notNull(),
27
+
ingredients: text('ingredients', { mode: 'json' }).$type<Partial<Ingredient>[]>().notNull(),
28
+
steps: text('steps', { mode: 'json' }).$type<Partial<Step>[]>().notNull(),
29
29
createdAt: dateIsoText("created_at").notNull(),
30
30
authorDid: did("author_did").notNull(),
31
31
}, t => ({
+51
-5
apps/api/src/ingest.ts
+51
-5
apps/api/src/ingest.ts
···
2
2
import { WebSocket } from "ws";
3
3
import { ingestLogger } from "./logger.js";
4
4
import env from "./config/env.js";
5
+
import { RecipeCollection, RecipeRecord } from "@cookware/lexicons";
6
+
import { db } from "./db/index.js";
7
+
import { recipeTable } from "./db/schema.js";
8
+
import { parseDid } from "./util/did.js";
9
+
import { and, eq } from "drizzle-orm";
5
10
6
11
export const newIngester = () => {
7
12
const jetstream = new Jetstream({
···
11
16
cursor: 0,
12
17
});
13
18
14
-
jetstream.onCreate("moe.hayden.cookware.recipe", event => {
15
-
ingestLogger.info(`New post: ${event.commit.record.title} (${event.commit.rkey})`);
16
-
});
19
+
jetstream.on("commit", async event => {
20
+
if (event.commit.operation == 'create' || event.commit.operation == 'update') {
21
+
const now = new Date();
22
+
const { record } = event.commit;
17
23
18
-
jetstream.onUpdate("moe.hayden.cookware.recipe", event => {
19
-
ingestLogger.info(`Updated post: ${event.commit.record.title} (${event.commit.rkey})`);
24
+
if (
25
+
event.commit.collection == RecipeCollection
26
+
&& record.$type == RecipeCollection
27
+
&& RecipeRecord.safeParse(record).success
28
+
) {
29
+
const res = await db
30
+
.insert(recipeTable)
31
+
.values({
32
+
rkey: event.commit.rkey,
33
+
title: record.title,
34
+
description: record.description,
35
+
ingredients: record.ingredients,
36
+
steps: record.steps,
37
+
authorDid: parseDid(event.did)!,
38
+
createdAt: now,
39
+
})
40
+
.onConflictDoUpdate({
41
+
target: recipeTable.id,
42
+
set: {
43
+
title: record.title,
44
+
description: record.description,
45
+
ingredients: record.ingredients,
46
+
steps: record.steps,
47
+
},
48
+
})
49
+
.execute();
50
+
51
+
ingestLogger.info({ res }, 'recipe ingested');
52
+
}
53
+
} else if (event.commit.operation == 'delete') {
54
+
const res = await db
55
+
.delete(recipeTable)
56
+
.where(
57
+
and(
58
+
eq(recipeTable.authorDid, parseDid(event.did)!),
59
+
eq(recipeTable.rkey, event.commit.rkey),
60
+
)
61
+
)
62
+
.execute();
63
+
64
+
ingestLogger.info({ res }, 'recipe deleted');
65
+
}
20
66
});
21
67
22
68
jetstream.on('open', () => {
+6
apps/api/src/recipes/index.ts
+6
apps/api/src/recipes/index.ts