Enhance error handling with invariant checks in comment, post, and vo…

…te APIs; add consumed offset check in receive hook

packages/frontpage/app/api/receive_hook/route.ts

@@ -6,6 +6,7 @@ import * as atprotoComment from "@/lib/data/atproto/comment";
 import * as atprotoVote from "@/lib/data/atproto/vote";
 import { getPdsUrl } from "@/lib/data/atproto/did";
 import { handleComment, handlePost, handleVote } from "@/lib/api/relayHandler";
+import { eq } from "drizzle-orm";
 
 export async function POST(request: Request) {
   const auth = request.headers.get("Authorization");
@@ -20,11 +21,22 @@ export async function POST(request: Request) {
   }
 
   const { ops, repo, seq } = commit.data;
+  const row = await db
+    .select()
+    .from(schema.ConsumedOffset)
+    .where(eq(schema.ConsumedOffset.offset, seq))
+    .limit(1);
+
+  const operationConsumed = Boolean(row[0]);
+  if (operationConsumed) {
+    console.log("Already consumed sequence:", seq);
+    return new Response("OK");
+  }
+
   const service = await getPdsUrl(repo);
   if (!service) {
     throw new Error("No AtprotoPersonalDataServer service found");
   }
-  console.log("ops", ops);
   const promises = ops.map(async (op) => {
     const { collection, rkey } = op.path;
     console.log("Processing", collection, rkey, op.action);
@@ -45,7 +57,6 @@ export async function POST(request: Request) {
   });
 
   await Promise.all(promises);
-  console.log("offset", seq);
   await db.insert(schema.ConsumedOffset).values({ offset: seq });
   return new Response("OK");
 }

packages/frontpage/lib/api/comment.ts

@@ -5,6 +5,7 @@ import { ensureUser } from "../data/user";
 import * as db from "../data/db/comment";
 import { DID } from "../data/atproto/did";
 import { createNotification } from "../data/db/notification";
+import { invariant } from "../utils";
 
 export type ApiCreateCommentInput = atproto.CommentInput & {
   repo: DID;
@@ -25,41 +26,36 @@ export async function createComment({
       content,
     });
 
-    if (!rkey || !cid) {
-      throw new DataLayerError("Failed to create comment");
-    }
+    invariant(rkey && cid, "Failed to create comment, rkey/cid missing");
 
     const comment = await atproto.getComment({
       rkey,
       repo: user.did,
     });
 
-    if (!comment) {
-      throw new DataLayerError(
-        "Failed to retrieve atproto comment, database creation aborted",
-      );
-    }
+    invariant(
+      comment,
+      "Failed to retrieve atproto comment, database creation aborted",
+    );
 
-    const createdComment = await db.createComment({
+    const dbCreatedComment = await db.createComment({
       cid,
       comment,
       repo,
       rkey: rkey,
     });
 
-    if (!createdComment) {
-      throw new DataLayerError("Failed to insert comment in database");
-    }
+    invariant(dbCreatedComment, "Failed to insert comment in database");
 
-    const didToNotify = createdComment.parent
-      ? createdComment.parent.authorDid
-      : createdComment.post.authordid;
+    const didToNotify = dbCreatedComment.parent
+      ? dbCreatedComment.parent.authorDid
+      : dbCreatedComment.post.authordid;
 
     if (didToNotify !== repo) {
       await createNotification({
-        commentId: createdComment.id,
+        commentId: dbCreatedComment.id,
         did: didToNotify,
-        reason: createdComment.parent ? "commentReply" : "postComment",
+        reason: dbCreatedComment.parent ? "commentReply" : "postComment",
       });
     }
   } catch (e) {

packages/frontpage/lib/api/post.ts

@@ -4,6 +4,7 @@ import * as atproto from "../data/atproto/post";
 import { ensureUser, getBlueskyProfile } from "../data/user";
 import { DataLayerError } from "../data/error";
 import { sendDiscordMessage } from "../discord";
+import { invariant } from "../utils";
 
 export interface ApiCreatePostInput extends Omit<atproto.Post, "createdAt"> {}
 
@@ -16,20 +17,17 @@ export async function createPost({ title, url }: ApiCreatePostInput) {
       url: url,
     });
 
-    if (!rkey || !cid) {
-      throw new DataLayerError("Failed to create post");
-    }
+    invariant(rkey && cid, "Failed to create post, rkey/cid missing");
 
     const post = await atproto.getPost({
       rkey,
       repo: user.did,
     });
 
-    if (!post) {
-      throw new DataLayerError(
-        "Failed to retrieve atproto post, database creation aborted",
-      );
-    }
+    invariant(
+      post,
+      "Failed to retrieve atproto post, database creation aborted",
+    );
 
     const dbCreatedPost = await db.createPost({
       post,
@@ -38,9 +36,7 @@ export async function createPost({ title, url }: ApiCreatePostInput) {
       cid,
     });
 
-    if (!dbCreatedPost) {
-      throw new DataLayerError("Failed to insert post in database");
-    }
+    invariant(dbCreatedPost, "Failed to insert post in database");
 
     const bskyProfile = await getBlueskyProfile(user.did);
 

packages/frontpage/lib/api/vote.ts

@@ -6,6 +6,7 @@ import { ensureUser } from "../data/user";
 import { PostCollection } from "../data/atproto/post";
 import { DID } from "../data/atproto/did";
 import { CommentCollection } from "../data/atproto/comment";
+import { invariant } from "../utils";
 
 export type ApiCreateVoteInput = {
   subjectRkey: string;
@@ -30,17 +31,14 @@ export async function createVote({
       subjectAuthorDid,
     });
 
-    if (!rkey || !cid) {
-      throw new DataLayerError("Failed to create vote");
-    }
+    invariant(rkey && cid, "Failed to create vote, rkey/cid missing");
 
     const vote = await atproto.getVote({ rkey, repo: user.did });
 
-    if (!vote) {
-      throw new DataLayerError(
-        "Failed to retrieve atproto vote, database creation aborted",
-      );
-    }
+    invariant(
+      vote,
+      "Failed to retrieve atproto vote, database creation aborted",
+    );
 
     if (subjectCollection == PostCollection) {
       const dbCreatedVote = await db.createPostVote({
@@ -50,9 +48,7 @@ export async function createVote({
         vote,
       });
 
-      if (!dbCreatedVote) {
-        throw new DataLayerError("Failed to insert post vote in database");
-      }
+      invariant(dbCreatedVote, "Failed to insert post vote in database");
     } else if (subjectCollection == CommentCollection) {
       const dbCreatedVote = await db.createCommentVote({
         repo: user.did,
@@ -61,9 +57,7 @@ export async function createVote({
         vote,
       });
 
-      if (!dbCreatedVote) {
-        throw new DataLayerError("Failed to insert comment vote in database");
-      }
+      invariant(dbCreatedVote, "Failed to insert post vote in database");
     }
   } catch (e) {
     throw new DataLayerError(`Failed to create post vote: ${e}`);