Storing Webhook Data

CREATE TABLE keyword_snapshots (
  id SERIAL PRIMARY KEY,
  project_id TEXT NOT NULL,
  keyword_id TEXT NOT NULL,
  keyword TEXT NOT NULL,
  desktop_position INTEGER,
  mobile_position INTEGER,
  ai_mode_position INTEGER,
  maps_position INTEGER,
  net_change INTEGER,
  location TEXT,
  recorded_at TIMESTAMP DEFAULT NOW()
);

CREATE INDEX idx_snapshots_project ON keyword_snapshots(project_id, recorded_at);
CREATE INDEX idx_snapshots_keyword ON keyword_snapshots(keyword_id, recorded_at);

CREATE TABLE webhook_events (
  id SERIAL PRIMARY KEY,
  event_type TEXT NOT NULL,
  project_id TEXT NOT NULL,
  payload JSONB NOT NULL,
  signature TEXT,
  processed_at TIMESTAMP DEFAULT NOW(),
  status TEXT DEFAULT 'received'
);

CREATE INDEX idx_events_type ON webhook_events(event_type, processed_at);

app.post('/webhooks/ranked', express.raw({ type: 'application/json' }), async (req, res) => {
  const signature = req.headers['x-webhook-signature'];
  const payload = req.body.toString();

  // Verify signature
  if (!verifyWebhook(payload, signature, process.env.RANKED_WEBHOOK_SECRET)) {
    return res.status(401).send();
  }

  const event = JSON.parse(payload);

  // 1. Store raw event immediately
  await db.query(
    'INSERT INTO webhook_events (event_type, project_id, payload, signature) VALUES ($1, $2, $3, $4)',
    [event.event, event.project_id, event, signature]
  );

  // 2. Return 200 quickly
  res.status(200).send('OK');

  // 3. Process asynchronously
  try {
    await processEvent(event);
    await db.query("UPDATE webhook_events SET status = 'processed' WHERE payload->>'timestamp' = $1", [event.timestamp]);
  } catch (err) {
    await db.query("UPDATE webhook_events SET status = 'failed' WHERE payload->>'timestamp' = $1", [event.timestamp]);
    console.error('Failed to process event:', err);
  }
});

async function processEvent(event) {
  // Check if already processed
  const existing = await db.query(
    "SELECT id FROM webhook_events WHERE event_type = $1 AND payload->>'timestamp' = $2 AND status = 'processed'",
    [event.event, event.timestamp]
  );

  if (existing.rows.length > 0) {
    console.log('Duplicate event, skipping');
    return;
  }

  // Process the event...
}