1  /**
  2   * Tests for replies pipeline stage
  3   *
  4   * Tests runRepliesStage(), getRepliesStats(), and processOptOuts()
  5   * using pg-mock (in-memory SQLite via db.js mock).
  6   */
  7  
  8  import { test, describe, mock, beforeEach } from 'node:test';
  9  import assert from 'node:assert/strict';
 10  import Database from 'better-sqlite3';
 11  import { createPgMock } from '../helpers/pg-mock.js';
 12  
 13  // ─── Create in-memory test DB ─────────────────────────────────────────────────
 14  
 15  const db = new Database(':memory:');
 16  
 17  db.exec(`
 18    CREATE TABLE sites (
 19      id INTEGER PRIMARY KEY AUTOINCREMENT,
 20      domain TEXT NOT NULL,
 21      landing_page_url TEXT,
 22      status TEXT DEFAULT 'found',
 23      keyword TEXT,
 24      score REAL,
 25      conversation_status TEXT,
 26      created_at TEXT DEFAULT CURRENT_TIMESTAMP,
 27      updated_at TEXT DEFAULT CURRENT_TIMESTAMP,
 28      rescored_at DATETIME
 29    );
 30  
 31    CREATE TABLE messages (
 32      id INTEGER PRIMARY KEY AUTOINCREMENT,
 33      site_id INTEGER,
 34      contact_method TEXT,
 35      contact_uri TEXT,
 36      direction TEXT NOT NULL DEFAULT 'outbound',
 37      approval_status TEXT,
 38      delivery_status TEXT,
 39      message_body TEXT,
 40      intent TEXT,
 41      sentiment TEXT,
 42      processed_at TEXT,
 43      created_at TEXT DEFAULT CURRENT_TIMESTAMP,
 44      updated_at TEXT DEFAULT CURRENT_TIMESTAMP,
 45      message_type TEXT DEFAULT 'outreach',
 46      raw_payload TEXT,
 47      read_at TEXT
 48    );
 49  
 50    CREATE TABLE opt_outs (
 51      id INTEGER PRIMARY KEY AUTOINCREMENT,
 52      phone TEXT,
 53      email TEXT,
 54      method TEXT NOT NULL CHECK(method IN ('sms', 'email')),
 55      opted_out_at DATETIME DEFAULT CURRENT_TIMESTAMP,
 56      source TEXT DEFAULT 'inbound',
 57      created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
 58      UNIQUE(phone, method),
 59      UNIQUE(email, method)
 60    );
 61  `);
 62  
 63  // ─── Mock db.js BEFORE importing replies.js ──────────────────────────────────
 64  
 65  mock.module('../../src/utils/db.js', {
 66    namedExports: createPgMock(db),
 67  });
 68  
 69  mock.module('../../src/utils/logger.js', {
 70    defaultExport: class {
 71      info() {}
 72      warn() {}
 73      error() {}
 74      success() {}
 75      debug() {}
 76    },
 77  });
 78  
 79  mock.module('../../src/utils/summary-generator.js', {
 80    namedExports: {
 81      generateStageCompletion: () => {},
 82      displayProgress: () => {},
 83    },
 84  });
 85  
 86  // Import AFTER mock.module
 87  const { runRepliesStage, getRepliesStats, processOptOuts } =
 88    await import('../../src/stages/replies.js');
 89  
 90  // ─── Helpers ─────────────────────────────────────────────────────────────────
 91  
 92  let siteSeq = 1;
 93  
 94  function clearTables() {
 95    db.prepare('DELETE FROM messages').run();
 96    db.prepare('DELETE FROM sites').run();
 97    db.prepare('DELETE FROM opt_outs').run();
 98    siteSeq = 1;
 99  }
100  
101  function insertSite() {
102    const id = siteSeq++;
103    db.prepare(
104      `INSERT INTO sites (id, domain, landing_page_url, status, keyword)
105       VALUES (?, ?, ?, 'outreach_sent', 'test kw')`
106    ).run(id, `site${id}.com`, `https://site${id}.com`);
107    return id;
108  }
109  
110  function insertOutreach(siteId, method = 'email') {
111    const result = db.prepare(
112      `INSERT INTO messages (site_id, contact_method, contact_uri, direction, approval_status, delivery_status, message_body)
113       VALUES (?, ?, 'test@example.com', 'outbound', 'approved', 'sent', 'Test proposal text')`
114    ).run(siteId, method);
115    return result.lastInsertRowid;
116  }
117  
118  function insertConversation(siteId, _status = 'awaiting_classification', contact_method = 'email') {
119    db.prepare(`UPDATE sites SET conversation_status = 'active' WHERE id = ?`).run(siteId);
120    const result = db.prepare(
121      `INSERT INTO messages (site_id, contact_uri, contact_method, message_body, direction)
122       VALUES (?, 'sender@test.com', ?, 'Hello!', 'inbound')`
123    ).run(siteId, contact_method);
124    return result.lastInsertRowid;
125  }
126  
127  // ─── Tests ───────────────────────────────────────────────────────────────────
128  
129  describe('Replies Stage', () => {
130    beforeEach(() => clearTables());
131  
132    describe('runRepliesStage', () => {
133      test('returns zero stats when no replies', async () => {
134        const result = await runRepliesStage();
135        assert.strictEqual(result.processed, 0);
136        assert.strictEqual(result.succeeded, 0);
137      });
138  
139      test('returns correct shape', async () => {
140        const result = await runRepliesStage();
141        assert.ok('processed' in result);
142        assert.ok('succeeded' in result);
143        assert.ok('failed' in result || result.failed === undefined);
144        assert.ok(typeof result.duration === 'number');
145      });
146  
147      test('processes awaiting_classification conversations', async () => {
148        const siteId = insertSite();
149        insertOutreach(siteId);
150        insertConversation(siteId, 'awaiting_classification');
151        insertConversation(siteId, 'awaiting_classification');
152  
153        const result = await runRepliesStage();
154        assert.ok(result.processed >= 2, `Expected >= 2 processed, got ${result.processed}`);
155      });
156  
157      test('respects limit option', async () => {
158        const siteId = insertSite();
159        insertOutreach(siteId);
160        for (let i = 0; i < 5; i++) {
161          insertConversation(siteId, 'awaiting_classification');
162        }
163  
164        const result = await runRepliesStage({ limit: 2 });
165        assert.ok(
166          result.processed <= 2,
167          `Should process at most 2 with limit, got ${result.processed}`
168        );
169      });
170  
171      test('showAll option includes processed conversations', async () => {
172        const siteId = insertSite();
173        insertOutreach(siteId);
174        insertConversation(siteId, 'awaiting_classification');
175  
176        const withAll = await runRepliesStage({ showAll: true });
177        const withoutAll = await runRepliesStage({ showAll: false });
178        assert.ok(withAll.processed >= withoutAll.processed);
179      });
180    });
181  
182    describe('getRepliesStats', () => {
183      test('returns stats object with processed_at column present', async () => {
184        const stats = await getRepliesStats();
185        assert.ok(typeof stats === 'object');
186        assert.ok('total_replies' in stats || 'processed' in stats);
187      });
188    });
189  
190    describe('processOptOuts', () => {
191      test('returns processed count', async () => {
192        const result = await processOptOuts();
193        assert.ok(typeof result === 'number' || typeof result === 'object');
194      });
195  
196      test('processes conversations with unsubscribe intent', async () => {
197        const siteId = insertSite();
198        insertOutreach(siteId);
199  
200        db.prepare(`UPDATE sites SET conversation_status = 'active' WHERE id = ?`).run(siteId);
201        db.prepare(
202          `INSERT INTO messages (site_id, contact_uri, contact_method, message_body, intent, direction)
203           VALUES (?, 'stop@example.com', 'sms', 'STOP', 'opt-out', 'inbound')`
204        ).run(siteId);
205  
206        await assert.doesNotReject(() => processOptOuts());
207      });
208  
209      test('does not throw when no conversations', async () => {
210        await assert.doesNotReject(() => processOptOuts());
211      });
212  
213      test('actually processes opt-out when sentiment is negative', async () => {
214        const siteId = insertSite();
215  
216        db.prepare(`UPDATE sites SET conversation_status = 'active' WHERE id = ?`).run(siteId);
217        db.prepare(
218          `INSERT INTO messages (site_id, contact_uri, contact_method, message_body, intent, sentiment, direction)
219           VALUES (?, 'stop@optin.com', 'sms', 'STOP please', 'opt-out', 'negative', 'inbound')`
220        ).run(siteId);
221  
222        const count = await processOptOuts();
223        assert.equal(count, 1, 'Should process 1 opt-out');
224  
225        const site = db.prepare('SELECT conversation_status FROM sites WHERE id = ?').get(siteId);
226        assert.equal(site.conversation_status, 'unsubscribed');
227      });
228    });
229  
230    describe('getRepliesStats intent breakdown', () => {
231      test('returns byIntent map with populated data', async () => {
232        const siteId = insertSite();
233        db.prepare(
234          `INSERT INTO messages (site_id, contact_method, contact_uri, direction, intent, sentiment, message_body)
235           VALUES (?, 'sms', '+1234', 'inbound', 'interested', 'positive', 'Yes I want it')`
236        ).run(siteId);
237        db.prepare(
238          `INSERT INTO messages (site_id, contact_method, contact_uri, direction, intent, sentiment, message_body)
239           VALUES (?, 'sms', '+5678', 'inbound', 'inquiry', 'neutral', 'What is this?')`
240        ).run(siteId);
241  
242        const stats = await getRepliesStats();
243        assert.ok(stats.byIntent, 'Should have byIntent map');
244        assert.ok(typeof stats.byIntent === 'object', 'byIntent should be an object');
245      });
246    });
247  });