db migration and perf improvemnet

backend/src/index.ts

@@ -5,10 +5,10 @@ import dotenv from "dotenv";
 import { Elysia } from "elysia";
 import { helmet } from "elysia-helmet";
 import path from "path";
-import { DistributionService } from "services/distribution/distribution.service";
 import configService, { validateEnv } from "./config/config";
 import RssPlugin from "./external/rss";
 import { db } from "./services/db";
+import { DistributionService } from "./services/distribution/distribution.service";
 import { SubmissionService } from "./services/submissions/submission.service";
 import { TwitterService } from "./services/twitter/client";
 import {
@@ -92,6 +92,7 @@ export async function main() {
         }),
       )
       .use(swagger())
+      .get("/health", () => new Response("OK", { status: 200 }))
       // API Routes
       .get("/api/last-tweet-id", () => {
         const lastTweetId = twitterService.getLastCheckedTweetId();
@@ -111,51 +112,40 @@ export async function main() {
           return { success: true };
         },
       )
-      .get("/api/submissions", ({ query }: { query: { status?: string } }) => {
-        const status = query?.status as
-          | "pending"
-          | "approved"
-          | "rejected"
-          | null;
-        return status
-          ? db.getSubmissionsByStatus(status)
-          : db.getAllSubmissions();
-      })
       .get(
-        "/api/submissions/:hashtagId",
-        ({ params: { hashtagId } }: { params: { hashtagId: string } }) => {
-          const config = configService.getConfig();
-          const feed = config.feeds.find((f) => f.id === hashtagId);
-          if (!feed) {
-            throw new Error(`Feed not found: ${hashtagId}`);
+        "/api/submission/:submissionId",
+        ({ params: { submissionId } }: { params: { submissionId: string } }) => {
+          const content = db.getSubmission(submissionId);
+          if (!content) {
+            throw new Error(`Content not found: ${submissionId}`);
           }
-          // this should be pending submissions
-          return db.getSubmissionsByFeed(hashtagId);
+          return content;
         },
       )
+      .get("/api/submissions", () => {
+        return db.getAllSubmissions();
+      })
       .get(
-        "/api/feed/:hashtagId",
-        ({ params: { hashtagId } }: { params: { hashtagId: string } }) => {
+        "/api/submissions/:feedId",
+        ({ params: { feedId } }: { params: { feedId: string } }) => {
           const config = configService.getConfig();
-          const feed = config.feeds.find((f) => f.id === hashtagId);
+          const feed = config.feeds.find((f) => f.id === feedId);
           if (!feed) {
-            throw new Error(`Feed not found: ${hashtagId}`);
+            throw new Error(`Feed not found: ${feedId}`);
           }
-
-          return db.getSubmissionsByFeed(hashtagId);
+          return db.getSubmissionsByFeed(feedId);
         },
       )
-      .get("/api/approved", () => {
-        return db.getSubmissionsByStatus("approved");
-      })
       .get(
-        "/api/content/:contentId",
-        ({ params: { contentId } }: { params: { contentId: string } }) => {
-          const content = db.getContent(contentId);
-          if (!content) {
-            throw new Error(`Content not found: ${contentId}`);
+        "/api/feed/:feedId",
+        ({ params: { feedId } }: { params: { feedId: string } }) => {
+          const config = configService.getConfig();
+          const feed = config.feeds.find((f) => f.id === feedId);
+          if (!feed) {
+            throw new Error(`Feed not found: ${feedId}`);
           }
-          return content;
+
+          return db.getSubmissionsByFeed(feedId);
         },
       )
       .get("/api/feeds", () => {
@@ -219,7 +209,10 @@ export async function main() {
           // Get approved submissions for this feed
           const submissions = db
             .getSubmissionsByFeed(feedId)
-            .filter((sub) => sub.status === "approved");
+            .filter((sub) =>
+              db.getFeedsBySubmission(sub.tweetId)
+                .some(feed => feed.status === "approved")
+            );
 
           if (submissions.length === 0) {
             return { processed: 0 };
@@ -289,7 +282,7 @@ export async function main() {
       "env",
       "twitter-init",
       "distribution-init",
-      "twitter-mentions",
+      "submission-monitor",
       "server",
     ].forEach((key) => {
       failSpinner(key, `Failed during ${key}`);

backend/src/services/db/index.ts

@@ -78,8 +78,8 @@ export class DatabaseService {
     queries.incrementDailySubmissionCount(this.db, userId).run();
   }
 
-  upsertFeed(feed: { id: string; name: string; description?: string }): void {
-    queries.upsertFeed(this.db, feed).run();
+  upsertFeeds(feeds: { id: string; name: string; description?: string }[]): void {
+    queries.upsertFeeds(this.db, feeds);
   }
 
   saveSubmissionToFeed(
@@ -98,15 +98,10 @@ export class DatabaseService {
     queries.removeFromSubmissionFeed(this.db, submissionId, feedId).run();
   }
 
-  getSubmissionsByFeed(feedId: string): TwitterSubmission[] {
+  getSubmissionsByFeed(feedId: string): (TwitterSubmission & { status: SubmissionStatus })[] {
     return queries.getSubmissionsByFeed(this.db, feedId);
   }
 
-  getContent(contentId: string): TwitterSubmission | null {
-    // For now, content is the same as submission since we're dealing with tweets
-    return this.getSubmission(contentId);
-  }
-
   // Feed Plugin Management
   getFeedPlugin(feedId: string, pluginId: string) {
     return queries.getFeedPlugin(this.db, feedId, pluginId);
@@ -144,21 +139,7 @@ export class DatabaseService {
 
   // Twitter Cache Management
   setTwitterCacheValue(key: string, value: string): void {
-    try {
-      twitterQueries.setTwitterCacheValue(this.db, key, value).run();
-    } catch (error: any) {
-      // Ignore write errors on read-only replicas
-      if (
-        error.code === "SQLITE_READONLY_DIRECTORY" ||
-        error.message?.includes("readonly database")
-      ) {
-        logger.info(
-          `Skipping Twitter cache write on read-only replica for key: ${key}`,
-        );
-        return;
-      }
-      throw error;
-    }
+    twitterQueries.setTwitterCacheValue(this.db, key, value).run();
   }
 
   getTwitterCacheValue(key: string): string | null {

backend/src/services/db/migrations/0002_schema_updates.sql

@@ -0,0 +1,76 @@
+-- Drop existing indexes first
+DROP INDEX IF EXISTS `submissions_user_id_idx`;--> statement-breakpoint
+DROP INDEX IF EXISTS `submissions_submitted_at_idx`;--> statement-breakpoint
+DROP INDEX IF EXISTS `submissions_status_idx`;--> statement-breakpoint
+DROP INDEX IF EXISTS `submissions_acknowledgment_idx`;--> statement-breakpoint
+DROP INDEX IF EXISTS `submissions_acknowledgment_tweet_id_unique`;--> statement-breakpoint
+
+-- Create new submissions table with updated schema
+PRAGMA foreign_keys=OFF;--> statement-breakpoint
+CREATE TABLE `__new_submissions` (
+	`tweet_id` text PRIMARY KEY NOT NULL,
+	`user_id` text NOT NULL,
+	`username` text NOT NULL,
+	`curator_id` text NOT NULL,
+	`curator_username` text NOT NULL,
+	`curator_tweet_id` text NOT NULL,
+	`content` text NOT NULL,
+	`curator_notes` text,
+	`submitted_at` text,
+	`created_at` text NOT NULL,
+	`updated_at` text
+);--> statement-breakpoint
+
+-- Copy data to new table
+INSERT INTO `__new_submissions` (
+    `tweet_id`,
+    `user_id`,
+    `username`,
+    `curator_id`,
+    `curator_username`,
+    `curator_tweet_id`,
+    `content`,
+    `curator_notes`,
+    `submitted_at`,
+    `created_at`,
+    `updated_at`
+)
+SELECT 
+    `tweet_id`,
+    `user_id`,
+    `username`,
+    `curator_id`,
+    `curator_username`,
+    COALESCE(`acknowledgment_tweet_id`, ''),
+    `content`,
+    `description`,
+    `submitted_at`,
+    `created_at`,
+    `updated_at`
+FROM `submissions`;--> statement-breakpoint
+
+-- Add columns to submission_feeds
+ALTER TABLE `submission_feeds` ADD COLUMN `status` text DEFAULT 'pending' NOT NULL;--> statement-breakpoint
+ALTER TABLE `submission_feeds` ADD COLUMN `moderation_response_tweet_id` text;--> statement-breakpoint
+
+-- Copy status from submissions to submission_feeds
+UPDATE submission_feeds 
+SET status = (
+    SELECT status 
+    FROM submissions 
+    WHERE submissions.tweet_id = submission_feeds.submission_id
+);--> statement-breakpoint
+
+-- Drop old submissions table and rename new one
+DROP TABLE `submissions`;--> statement-breakpoint
+ALTER TABLE `__new_submissions` RENAME TO `submissions`;--> statement-breakpoint
+
+-- Add feed_id to moderation_history
+ALTER TABLE `moderation_history` ADD COLUMN `feed_id` text NOT NULL REFERENCES feeds(id);--> statement-breakpoint
+
+-- Recreate indexes
+CREATE INDEX IF NOT EXISTS `submissions_user_id_idx` ON `submissions` (`user_id`);--> statement-breakpoint
+CREATE INDEX IF NOT EXISTS `submissions_submitted_at_idx` ON `submissions` (`submitted_at`);--> statement-breakpoint
+CREATE INDEX IF NOT EXISTS `moderation_history_feed_idx` ON `moderation_history` (`feed_id`);--> statement-breakpoint
+
+PRAGMA foreign_keys=ON;

backend/src/services/db/migrations/meta/_journal.json

@@ -1,6 +1,7 @@
 {
-  "version": "7",
+  "version": "5",
   "dialect": "sqlite",
+  "id": "sqlite",
   "entries": [
     {
       "idx": 0,
@@ -19,15 +20,8 @@
     {
       "idx": 2,
       "version": "6",
-      "when": 1737162664131,
-      "tag": "0002_sweet_ben_grimm",
-      "breakpoints": true
-    },
-    {
-      "idx": 3,
-      "version": "6",
       "when": 1737163062577,
-      "tag": "0003_melted_redwing",
+      "tag": "0002_schema_updates",
       "breakpoints": true
     }
   ]

backend/src/services/db/operations.ts

@@ -6,6 +6,7 @@ import {
   deleteOldRssItems,
 } from "../rss/queries";
 
+// These are made available for plugins
 export class DBOperations {
   constructor(private db: BunSQLiteDatabase) {}