fix: resolve issues #73, #75, #79 (indexes, notifications, event filtering)

database/schema.sql

@@ -76,3 +76,34 @@ CREATE TABLE IF NOT EXISTS stream_tags (
 
 CREATE INDEX IF NOT EXISTS idx_stream_tags_stream_id ON stream_tags(stream_id);
 CREATE INDEX IF NOT EXISTS idx_stream_tags_tag_id    ON stream_tags(tag_id);
+
+-- ---------------------------------------------------------------------
+-- Issue #73: Indexes for common query patterns
+-- Rollback: DROP INDEX idx_streams_user_id_status, idx_stream_events_stream_id_occurred_at, idx_users_email;
+-- ---------------------------------------------------------------------
+
+CREATE INDEX IF NOT EXISTS idx_streams_user_id_status
+    ON streams(user_id, status);
+
+CREATE INDEX IF NOT EXISTS idx_stream_events_stream_id_occurred_at
+    ON stream_events(stream_id, created_at DESC);
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_users_email
+    ON users(email);
+
+-- ---------------------------------------------------------------------
+-- Issue #75: Notifications table
+-- Rollback: DROP TABLE notifications;
+-- ---------------------------------------------------------------------
+
+CREATE TABLE IF NOT EXISTS notifications (
+    id         SERIAL PRIMARY KEY,
+    user_id    INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+    type       VARCHAR(100) NOT NULL,
+    payload    JSONB NOT NULL DEFAULT '{}',
+    read_at    TIMESTAMP,
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE INDEX IF NOT EXISTS idx_notifications_user_id ON notifications(user_id);
+CREATE INDEX IF NOT EXISTS idx_notifications_unread  ON notifications(user_id) WHERE read_at IS NULL;

xstreamroll-processing/src/pipeline/event-filter.ts

@@ -0,0 +1,38 @@
+import { StreamEvent } from "../session"
+
+export interface FilterConfig {
+  /** Event types to suppress. Events matching any entry are dropped silently. */
+  blockedEventTypes: string[]
+}
+
+/**
+ * Per-stream filter config store with hot-reload support.
+ *
+ * Configs are keyed by streamId. Calling `setConfig` replaces the
+ * config for that stream immediately — no worker restart required.
+ */
+export class EventFilter {
+  private readonly configs = new Map<string, FilterConfig>()
+
+  /** Update (or set) the filter config for a stream. */
+  setConfig(streamId: string, config: FilterConfig): void {
+    this.configs.set(streamId, { blockedEventTypes: [...config.blockedEventTypes] })
+  }
+
+  /** Remove the filter config for a stream (all events pass through). */
+  clearConfig(streamId: string): void {
+    this.configs.delete(streamId)
+  }
+
+  /**
+   * Returns `true` when the event should be forwarded to subscribers,
+   * `false` when it should be dropped silently.
+   */
+  allow(event: StreamEvent): boolean {
+    const config = this.configs.get(event.streamId)
+    if (!config) return true
+    const eventType = event.data["type"]
+    if (typeof eventType !== "string") return true
+    return !config.blockedEventTypes.includes(eventType)
+  }
+}

xstreamroll-processing/src/pipeline/index.ts

@@ -0,0 +1,2 @@
+export { EventFilter } from "./event-filter"
+export type { FilterConfig } from "./event-filter"

xstreamroll-processing/src/worker.ts

@@ -1,5 +1,6 @@
 import axios from "axios"
 import { env } from "./config"
+import { EventFilter } from "./pipeline"
 import { SessionRegistry } from "./session-registry"
 import { ProcessedStreamEvent, StreamEvent } from "./session"
 
@@ -21,6 +22,8 @@ const registry = new SessionRegistry(
   { maxConcurrentSessions: MAX_CONCURRENT_SESSIONS },
 )
 
+const filter = new EventFilter()
+
 async function pollOnce(): Promise<void> {
   let events: StreamEvent[] = []
   try {
@@ -37,6 +40,9 @@ async function pollOnce(): Promise<void> {
       console.warn(`[${WORKER_ID}] dropping malformed event`, event)
       continue
     }
+    if (!filter.allow(event)) {
+      continue // silently drop filtered events
+    }
     const result = registry.route(event)
     if (result === "capacity") {
       const cap = registry.capacity()