Receiving webhooks

import express from "express";
import { db } from "./db.js";
import { jobQueue } from "./queue.js";
 
const app = express();
app.use(express.json({ limit: "10mb" }));
 
app.post("/webhooks/cleanlist", async (req, res) => {
  const payload = req.body;
  const workflowId = payload?.workflow_id;
 
  if (!workflowId) {
    return res.status(400).json({ error: "missing workflow_id" });
  }
 
  // Rule 3: only runs we started
  const run = await db.workflows.findOne({ id: workflowId });
  if (!run) {
    return res.status(404).json({ error: "unknown workflow_id" });
  }
 
  // Rule 2: idempotency
  if (run.processed_at) {
    return res.status(200).end();
  }
 
  // Rule 1: save, answer, process later
  await db.webhookInbox.insert({
    workflow_id: workflowId,
    payload,
    received_at: new Date(),
  });
  await jobQueue.enqueue("process-cleanlist-results", { workflowId });
 
  res.status(200).end();
});
 
app.listen(3000, () => console.log("Receiver listening on :3000"));

export async function processCleanlistResults({ workflowId }) {
  const inbox = await db.webhookInbox.findOne({ workflow_id: workflowId });
  const { results, summary, status } = inbox.payload;
 
  const failedLeadIds = [];
  for (const row of results ?? []) {
    if (row.status === "completed") {
      await db.leads.upsert({
        where: { source_id: row.task_id },
        create: {
          source_id: row.task_id,
          email: row.result?.email,
          email_status: row.result?.email_status,
          phones: row.result?.phones ?? [],
          first_name: row.input?.first_name,
          last_name: row.input?.last_name,
          company_domain: row.input?.company_domain,
        },
        update: {
          email: row.result?.email,
          email_status: row.result?.email_status,
          phones: row.result?.phones ?? [],
        },
      });
    } else if (row.lead_id) {
      failedLeadIds.push(row.lead_id);
    }
  }
 
  // Rule 4: errors and throttled rows are worth re-enriching later.
  // Missed rows (nothing found) will not improve on a retry.
  if (summary.errors + summary.throttled > 0) {
    await flagForReEnrichment(workflowId, failedLeadIds);
  }
 
  await db.workflows.update(workflowId, {
    processed_at: new Date(),
    final_status: status,
    summary,
  });
}

from datetime import datetime, timezone
 
from flask import Flask, request, jsonify
from redis import Redis
from rq import Queue
 
from .db import db
from .jobs import process_cleanlist_results
 
app = Flask(__name__)
queue = Queue(connection=Redis())
 
@app.post("/webhooks/cleanlist")
def cleanlist_webhook():
    payload = request.get_json(silent=True) or {}
    workflow_id = payload.get("workflow_id")
 
    if not workflow_id:
        return jsonify(error="missing workflow_id"), 400
 
    run = db.workflows.find_one({"id": workflow_id})
    if not run:
        return jsonify(error="unknown workflow_id"), 404
 
    if run.get("processed_at"):
        return "", 200  # already handled
 
    db.webhook_inbox.insert_one({
        "workflow_id": workflow_id,
        "payload": payload,
        "received_at": datetime.now(timezone.utc),
    })
    queue.enqueue(process_cleanlist_results, workflow_id)
    return "", 200

import { SQSClient, SendMessageCommand } from "@aws-sdk/client-sqs";
 
const sqs = new SQSClient({});
 
export const handler = async (event) => {
  const payload = JSON.parse(event.body);
  const workflowId = payload?.workflow_id;
 
  if (!workflowId) {
    return { statusCode: 400, body: '{"error":"missing workflow_id"}' };
  }
 
  // Hand off to a FIFO queue and answer right away
  await sqs.send(
    new SendMessageCommand({
      QueueUrl: process.env.RESULTS_QUEUE_URL,
      MessageBody: event.body,
      MessageDeduplicationId: workflowId,
      MessageGroupId: "cleanlist",
    }),
  );
 
  return { statusCode: 200, body: "" };
};

deliveries = requests.get(
    "https://api.cleanlist.ai/api/v1/public/webhooks/deliveries",
    headers={"Authorization": f"Bearer {API_KEY}"},
    params={"workflow_id": workflow_id},
).json()
 
for d in deliveries:
    print(d["attempt_number"], d["status"], d["response_status_code"], d["error_message"])

status = requests.get(
    "https://api.cleanlist.ai/api/v1/public/enrich/status",
    headers={"Authorization": f"Bearer {API_KEY}"},
    params={"task_id": task_id},
).json()["task"]