Merge pull request #6 from BTreeMap/copilot/harden-persistence-layer

feat(store): durable jobs, outbox, and inbound dedup for restart-safe persistence

Author: MinecraftFuns

Makefile

@@ -5,4 +5,8 @@ build:
 test:
 	go test ./...
 
-.PHONY: build test
+test-postgres:
+	@echo "Run with: POSTGRES_DSN_TEST='postgres://user:pass@host:5432/dbname?sslmode=disable' make test-postgres"
+	POSTGRES_DSN_TEST="$${POSTGRES_DSN_TEST}" go test ./internal/store/... -v -count=1
+
+.PHONY: build test test-postgres

docker-compose.yml

@@ -0,0 +1,49 @@
+# docker-compose.yml - PromptPipe development/production stack
+#
+# Data persistence: All durable state is stored in named volumes.
+# "docker compose down && docker compose up -d" preserves all state.
+# Use "docker compose down -v" to explicitly destroy volumes.
+
+services:
+  promptpipe:
+    build:
+      context: .
+      dockerfile: docker/Dockerfile
+    ports:
+      - "${API_PORT:-8080}:8080"
+    environment:
+      - DATABASE_DSN=${DATABASE_DSN:-postgres://promptpipe:promptpipe@postgres:5432/promptpipe?sslmode=disable}
+      - WHATSAPP_DB_DSN=${WHATSAPP_DB_DSN:-file:/data/whatsmeow.db?_foreign_keys=on}
+      - OPENAI_API_KEY=${OPENAI_API_KEY:-}
+      - API_ADDR=:8080
+      - PROMPTPIPE_STATE_DIR=/data
+      - PROMPTPIPE_DEBUG=${PROMPTPIPE_DEBUG:-false}
+      - GENAI_MODEL=${GENAI_MODEL:-}
+      - GENAI_TEMPERATURE=${GENAI_TEMPERATURE:-0.1}
+      - AUTO_ENROLL_NEW_USERS=${AUTO_ENROLL_NEW_USERS:-false}
+      - AUTO_FEEDBACK_AFTER_PROMPT_ENABLED=${AUTO_FEEDBACK_AFTER_PROMPT_ENABLED:-true}
+    volumes:
+      - promptpipe-data:/data
+    depends_on:
+      postgres:
+        condition: service_healthy
+    restart: unless-stopped
+
+  postgres:
+    image: postgres:16-alpine
+    environment:
+      POSTGRES_USER: promptpipe
+      POSTGRES_PASSWORD: promptpipe
+      POSTGRES_DB: promptpipe
+    volumes:
+      - postgres-data:/var/lib/postgresql/data
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U promptpipe"]
+      interval: 5s
+      timeout: 5s
+      retries: 5
+    restart: unless-stopped
+
+volumes:
+  promptpipe-data:
+  postgres-data:

docs/persistence-audit.md

@@ -0,0 +1,82 @@
+# Persistence Audit – State Ledger
+
+This document inventories all state that influences user-visible behavior and
+must survive a `docker compose down && docker compose up -d` cycle.
+
+## State Items
+
+| # | State Item | Owner (package) | Must Survive Restart? | Current Storage | Proposed Durable Representation | Idempotency / Dedup Strategy |
+|---|---|---|---|---|---|---|
+| 1 | **In-memory timers** (`SimpleTimer.timers` map) | `flow/timer.go` | Yes | In-memory map of `*timerEntry` keyed by timer ID | `jobs` table (`kind` + `run_at` + `payload_json`) | `dedupe_key` unique constraint; on restart, requeue stale running jobs |
+| 2 | **Recurring schedule timers** (daily prompt schedule) | `flow/timer.go` | Yes | In-memory via `time.AfterFunc` with self-rescheduling | `jobs` table with `kind=recurring_prompt`; after execution, enqueue next occurrence | `dedupe_key` per participant + schedule; only one active job per schedule |
+| 3 | **Daily prompt pending state** (`dailyPromptPendingState`) | `flow/scheduler_tool.go` | Yes | Persisted in `flow_states.state_data` as JSON (key `daily_prompt_pending`) | Keep in `flow_states`; reminder becomes a `jobs` row instead of `time.AfterFunc` | Check pending state before sending reminder; clear after send |
+| 4 | **Pending state transition timer** (delayed `transition_state`) | `flow/state_transition_tool.go` | Yes | Timer ID stored in `flow_states.state_data`, actual timer in memory | `jobs` table with `kind=state_transition` | `dedupe_key` per participant + flow type; cancel = mark job canceled |
+| 5 | **Feedback follow-up timer** | `flow/feedback_module.go` | Yes | In-memory timer via `SimpleTimer` | `jobs` table with `kind=feedback_followup` | `dedupe_key` per participant; skip if participant state changed |
+| 6 | **Feedback initial timeout timer** | `flow/feedback_module.go` | Yes | In-memory timer via `SimpleTimer` | `jobs` table with `kind=feedback_timeout` | `dedupe_key` per participant; skip if already responded |
+| 7 | **LID-by-phone cache** (`lidByPhone` map) | `messaging/whatsapp_service.go` | No | In-memory map | Remains in-memory; repopulated on first message from each contact | N/A – cache miss just means first send uses phone number directly |
+| 8 | **Outgoing WhatsApp messages** (sent via `SendMessage`) | `messaging/whatsapp_service.go` | Yes | Fire-and-forget; receipt stored in `receipts` table | `outbox_messages` table; flow enqueues, sender worker delivers | `dedupe_key` prevents duplicate sends on restart |
+| 9 | **Inbound WhatsApp messages** (received via event handler) | `messaging/whatsapp_service.go` | Yes (dedup) | Processed immediately; response stored in `responses` table | `inbound_dedup` table keyed by WhatsApp message ID | Insert-or-skip on message ID; prevents reprocessing after restart |
+| 10 | **Flow state** (`flow_states` table) | `store/`, `flow/state_manager.go` | Yes | SQLite/Postgres `flow_states` table | Already persisted | N/A – already durable |
+| 11 | **Conversation participants** | `store/` | Yes | SQLite/Postgres `conversation_participants` table | Already persisted | N/A – already durable |
+| 12 | **Registered hooks** | `store/` | Yes | SQLite/Postgres `registered_hooks` table | Already persisted | N/A – already durable |
+| 13 | **Receipts / Responses** | `store/` | Yes | SQLite/Postgres tables | Already persisted | N/A – already durable |
+
+## New Tables
+
+### `jobs`
+
+Replaces all in-memory `time.AfterFunc` timers with durable, restart-safe job records.
+
+| Column | Type | Description |
+|---|---|---|
+| `id` | TEXT (UUID) | Primary key |
+| `kind` | TEXT | Job type (e.g., `recurring_prompt`, `state_transition`, `feedback_followup`, `daily_prompt_reminder`) |
+| `run_at` | TIMESTAMP | When the job should execute |
+| `payload_json` | TEXT/JSON | Job-specific parameters |
+| `status` | TEXT | `queued`, `running`, `done`, `failed`, `canceled` |
+| `attempt` | INTEGER | Current attempt number |
+| `max_attempts` | INTEGER | Maximum retry attempts |
+| `last_error` | TEXT | Last error message (nullable) |
+| `locked_at` | TIMESTAMP | When a worker claimed this job (nullable) |
+| `dedupe_key` | TEXT | Unique constraint for preventing duplicates (nullable) |
+| `created_at` | TIMESTAMP | Row creation time |
+| `updated_at` | TIMESTAMP | Last update time |
+
+### `outbox_messages`
+
+Ensures outgoing WhatsApp sends are restart-safe and idempotent.
+
+| Column | Type | Description |
+|---|---|---|
+| `id` | TEXT (UUID) | Primary key |
+| `participant_id` | TEXT | Target participant |
+| `kind` | TEXT | Message type (e.g., `prompt`, `reminder`, `feedback_followup`) |
+| `payload_json` | TEXT/JSON | Message content and metadata |
+| `status` | TEXT | `queued`, `sending`, `sent`, `failed` |
+| `attempts` | INTEGER | Send attempt count |
+| `next_attempt_at` | TIMESTAMP | When to retry (nullable) |
+| `dedupe_key` | TEXT | Unique constraint for preventing duplicate sends |
+| `locked_at` | TIMESTAMP | When a worker claimed this message (nullable) |
+| `last_error` | TEXT | Last error message (nullable) |
+| `created_at` | TIMESTAMP | Row creation time |
+| `updated_at` | TIMESTAMP | Last update time |
+
+### `inbound_dedup`
+
+Prevents reprocessing of inbound messages after restart.
+
+| Column | Type | Description |
+|---|---|---|
+| `message_id` | TEXT | Primary key – stable WhatsApp message ID |
+| `participant_id` | TEXT | Sender identifier |
+| `received_at` | TIMESTAMP | When the message was first seen |
+| `processed_at` | TIMESTAMP | When processing completed (nullable) |
+
+## Invariants
+
+1. Any state transition that implies future work must, in ONE DB transaction:
+   - Update `flow_states`
+   - Enqueue `jobs` and/or `outbox_messages`
+2. Timers are never used for durable behavior – they become jobs.
+3. `dedupe_key` constraints ensure restarts and retries do not duplicate work.
+4. On startup: requeue stale `running` jobs/outbox (locked_at older than threshold).

docs/state-persistence-recovery.md

@@ -101,6 +101,46 @@ CREATE TABLE conversation_participants (...);
 4. **Maintainable**: Business logic separated from infrastructure concerns
 5. **No Breaking Changes**: Uses existing database schema
 
+## Durable Jobs and Outbox (Persistence Hardening)
+
+In addition to the recovery system above, the persistence layer includes:
+
+### Durable Jobs (`internal/store/job_repo.go`, `job_runner.go`)
+
+All time-based behavior that must survive restarts is represented as **persisted job records** in the `jobs` table, replacing in-memory `time.AfterFunc` calls:
+
+- **`JobRepo`**: Interface for enqueueing, claiming, completing, failing, and canceling jobs. Implementations exist for both SQLite and Postgres.
+- **`JobRunner`**: Background worker that polls for due jobs and dispatches them to registered handlers. Supports exponential backoff on failure and crash recovery via `RequeueStaleRunningJobs()`.
+- **Dedupe keys**: Prevent duplicate jobs for the same intent (e.g., one scheduled prompt per participant per schedule).
+
+### Outbox Messages (`internal/store/outbox_repo.go`, `outbox_sender.go`)
+
+Outgoing WhatsApp messages are persisted to the `outbox_messages` table before being sent:
+
+- **`OutboxRepo`**: Interface for enqueueing and managing outbound messages. Dedupe keys prevent duplicate sends.
+- **`OutboxSender`**: Background worker that claims due messages and sends them via the messaging service. Supports retry with backoff.
+- On startup, stale `sending` messages are requeued.
+
+### Inbound Dedup (`internal/store/dedup_repo.go`)
+
+The `inbound_dedup` table prevents reprocessing of inbound WhatsApp messages after restart:
+
+- **`DedupRepo`**: Interface for recording and checking message IDs.
+- On message arrival, the system checks if the message ID was already processed and skips if so.
+
+### Startup Reconciliation
+
+On startup:
+1. Stale `running` jobs are requeued (`RequeueStaleRunningJobs`)
+2. Stale `sending` outbox messages are requeued (`RequeueStaleSendingMessages`)
+3. Existing recovery (response handlers, scheduler reminders) continues as before
+
+### Docker Compose Persistence
+
+The `docker-compose.yml` uses named volumes for both Postgres data and the PromptPipe state directory. `docker compose down && docker compose up -d` preserves all state. Only `docker compose down -v` destroys volumes.
+
+See `docs/persistence-audit.md` for the complete state ledger.
+
 ## Integration Example
 
 ```go

docs/storage.md

@@ -50,12 +50,37 @@ Tables used by the current Go service:
 | `flow_states` | Flow state + JSON state data |
 | `conversation_participants` | Enrolled conversation participants |
 | `registered_hooks` | Persisted response hooks |
+| `jobs` | Durable job queue (replaces in-memory timers) |
+| `outbox_messages` | Restart-safe outgoing message queue |
+| `inbound_dedup` | Inbound message deduplication |
 
 **Legacy tables** (present in migrations but unused by current code):
 
 - `intervention_participants`
 - `intervention_responses`
 
+## Durable jobs (`jobs` table)
+
+All time-based behavior (scheduled prompts, feedback timeouts, state transitions) is represented as persisted job records instead of in-memory timers. Each job has:
+
+- `kind` – handler type (e.g., `recurring_prompt`, `state_transition`, `feedback_followup`)
+- `run_at` – when the job should execute
+- `payload_json` – job-specific parameters
+- `status` – lifecycle state (`queued` → `running` → `done`/`failed`/`canceled`)
+- `dedupe_key` – prevents duplicate jobs for the same intent
+
+On startup, stale `running` jobs (locked before a configurable threshold) are requeued.
+
+## Outbox messages (`outbox_messages` table)
+
+Outgoing WhatsApp messages are first persisted to the outbox, then sent by a background worker. This ensures messages are not lost on crash. Each message has a `dedupe_key` to prevent duplicate sends.
+
+## Inbound dedup (`inbound_dedup` table)
+
+Inbound WhatsApp messages are deduplicated by message ID to prevent reprocessing after restart.
+
+See `docs/persistence-audit.md` for a full state ledger.
+
 ## Flow state data
 
 `flow_states.state_data` stores JSON (SQLite) or JSONB (Postgres) key/value pairs for conversation flow. Keys are defined in `internal/models/flow_types.go` (e.g., `conversationHistory`, `userProfile`, `scheduleRegistry`).

internal/store/dedup_postgres.go

@@ -0,0 +1,51 @@
+package store
+
+import (
+	"database/sql"
+	"fmt"
+	"time"
+)
+
+// Compile-time check that PostgresStore implements DedupRepo.
+var _ DedupRepo = (*PostgresStore)(nil)
+
+func (s *PostgresStore) IsDuplicate(messageID string) (bool, error) {
+	var id string
+	err := s.db.QueryRow(`SELECT message_id FROM inbound_dedup WHERE message_id = $1`, messageID).Scan(&id)
+	if err == sql.ErrNoRows {
+		return false, nil
+	}
+	if err != nil {
+		return false, fmt.Errorf("dedup check failed: %w", err)
+	}
+	return true, nil
+}
+
+func (s *PostgresStore) RecordInbound(messageID, participantID string) (bool, error) {
+	now := time.Now()
+	result, err := s.db.Exec(
+		`INSERT INTO inbound_dedup (message_id, participant_id, received_at) VALUES ($1, $2, $3) ON CONFLICT (message_id) DO NOTHING`,
+		messageID, participantID, now,
+	)
+	if err != nil {
+		return false, fmt.Errorf("record inbound failed: %w", err)
+	}
+
+	n, err := result.RowsAffected()
+	if err != nil {
+		return false, fmt.Errorf("dedup rows affected check failed: %w", err)
+	}
+	return n > 0, nil
+}
+
+func (s *PostgresStore) MarkProcessed(messageID string) error {
+	now := time.Now()
+	_, err := s.db.Exec(
+		`UPDATE inbound_dedup SET processed_at = $1 WHERE message_id = $2`,
+		now, messageID,
+	)
+	if err != nil {
+		return fmt.Errorf("mark processed failed: %w", err)
+	}
+	return nil
+}

internal/store/dedup_repo.go

@@ -0,0 +1,28 @@
+// Package store provides the DedupRepo interface for inbound message deduplication.
+package store
+
+import (
+	"time"
+)
+
+// DedupRecord represents an inbound message deduplication record.
+type DedupRecord struct {
+	MessageID     string     `json:"message_id"`
+	ParticipantID string     `json:"participant_id"`
+	ReceivedAt    time.Time  `json:"received_at"`
+	ProcessedAt   *time.Time `json:"processed_at"`
+}
+
+// DedupRepo defines the interface for inbound message deduplication.
+type DedupRepo interface {
+	// IsDuplicate checks if a message ID has already been processed.
+	// Returns true if the message was already seen.
+	IsDuplicate(messageID string) (bool, error)
+
+	// RecordInbound inserts a new inbound message record. Returns false if the
+	// message was already recorded (duplicate).
+	RecordInbound(messageID, participantID string) (bool, error)
+
+	// MarkProcessed sets the processed_at timestamp for a message.
+	MarkProcessed(messageID string) error
+}

internal/store/dedup_sqlite.go

@@ -0,0 +1,55 @@
+package store
+
+import (
+	"database/sql"
+	"fmt"
+	"time"
+)
+
+// Compile-time check that SQLiteStore implements DedupRepo.
+var _ DedupRepo = (*SQLiteStore)(nil)
+
+func (s *SQLiteStore) IsDuplicate(messageID string) (bool, error) {
+	var id string
+	err := s.db.QueryRow(`SELECT message_id FROM inbound_dedup WHERE message_id = ?`, messageID).Scan(&id)
+	if err == sql.ErrNoRows {
+		return false, nil
+	}
+	if err != nil {
+		return false, fmt.Errorf("dedup check failed: %w", err)
+	}
+	return true, nil
+}
+
+func (s *SQLiteStore) RecordInbound(messageID, participantID string) (bool, error) {
+	// First check if it already exists
+	exists, err := s.IsDuplicate(messageID)
+	if err != nil {
+		return false, err
+	}
+	if exists {
+		return false, nil
+	}
+
+	now := time.Now()
+	_, err = s.db.Exec(
+		`INSERT OR IGNORE INTO inbound_dedup (message_id, participant_id, received_at) VALUES (?, ?, ?)`,
+		messageID, participantID, now,
+	)
+	if err != nil {
+		return false, fmt.Errorf("record inbound failed: %w", err)
+	}
+	return true, nil
+}
+
+func (s *SQLiteStore) MarkProcessed(messageID string) error {
+	now := time.Now()
+	_, err := s.db.Exec(
+		`UPDATE inbound_dedup SET processed_at = ? WHERE message_id = ?`,
+		now, messageID,
+	)
+	if err != nil {
+		return fmt.Errorf("mark processed failed: %w", err)
+	}
+	return nil
+}