MongoDB Integration

/**
 * MongoDB Change Streams watcher for Certyo.
 * Watches a collection for insert/update/replace/delete events,
 * transforms them into Certyo records, and ingests via the API.
 *
 * Install: npm install mongodb node-fetch
 * Usage:   CERTYO_API_KEY=xxx CERTYO_TENANT_ID=yyy node certyo-mongo-watcher.js
 */

const { MongoClient } = require("mongodb");

const MONGO_URI = process.env.MONGO_URI || "mongodb://localhost:27017/?replicaSet=rs0";
const DATABASE = process.env.MONGO_DATABASE || "productdb";
const COLLECTION = process.env.MONGO_COLLECTION || "products";

const CERTYO_API_KEY = process.env.CERTYO_API_KEY;
const CERTYO_TENANT_ID = process.env.CERTYO_TENANT_ID;
const CERTYO_BASE_URL = process.env.CERTYO_BASE_URL || "https://www.certyos.com";

if (!CERTYO_API_KEY || !CERTYO_TENANT_ID) {
  console.error("CERTYO_API_KEY and CERTYO_TENANT_ID are required");
  process.exit(1);
}

async function main() {
  const client = new MongoClient(MONGO_URI);
  await client.connect();
  console.log("Connected to MongoDB");

  const db = client.db(DATABASE);
  const collection = db.collection(COLLECTION);
  const cursorsCollection = db.collection("_certyo_cursors");

  // Load persisted resume token (if any)
  const cursor = await cursorsCollection.findOne({ _id: COLLECTION });
  const resumeAfter = cursor?.resumeToken || undefined;

  const pipeline = [
    {
      $match: {
        operationType: { $in: ["insert", "update", "replace", "delete"] },
      },
    },
  ];

  const options = {
    fullDocument: "updateLookup",
    ...(resumeAfter && { resumeAfter }),
  };

  console.log(
    resumeAfter
      ? "Resuming Change Stream from saved token"
      : "Starting Change Stream from current position"
  );

  const changeStream = collection.watch(pipeline, options);

  changeStream.on("change", async (event) => {
    try {
      const record = mapChangeEvent(event);
      const result = await ingestRecord(record);

      if (result) {
        console.log(
          `Ingested ${record.recordId}: hash=${result.recordHash}`
        );
      }

      // Persist resume token after successful ingestion
      await cursorsCollection.updateOne(
        { _id: COLLECTION },
        {
          $set: {
            resumeToken: event._id,
            lastProcessedAt: new Date(),
            lastRecordId: record.recordId,
          },
        },
        { upsert: true }
      );
    } catch (err) {
      console.error("Error processing change event:", err.message);
    }
  });

  changeStream.on("error", (err) => {
    console.error("Change Stream error:", err.message);
    // The driver will attempt to resume automatically using the resume token
  });

  // Graceful shutdown
  process.on("SIGINT", async () => {
    console.log("Shutting down...");
    await changeStream.close();
    await client.close();
    process.exit(0);
  });
}

function mapChangeEvent(event) {
  const operationType =
    event.operationType === "replace" ? "update" : event.operationType;

  // For deletes, fullDocument is null — use documentKey
  const doc = event.fullDocument || {};
  const docId =
    event.documentKey?._id?.toString() || doc._id?.toString() || "unknown";

  return {
    tenantId: CERTYO_TENANT_ID,
    database: DATABASE,
    collection: COLLECTION,
    recordId: docId,
    recordVersion: event._id?._data || "1",
    operationType,
    recordPayload:
      operationType === "delete"
        ? { _id: docId, _deleted: true }
        : sanitizeDocument(doc),
    sourceTimestamp:
      event.clusterTime?.toNumber()
        ? new Date(event.clusterTime.toNumber() * 1000).toISOString()
        : new Date().toISOString(),
    idempotencyKey: `mongo-${COLLECTION}-${docId}-${event._id?._data?.substring(0, 16) || Date.now()}`,
  };
}

function sanitizeDocument(doc) {
  // Convert MongoDB-specific types to plain JSON
  const sanitized = { ...doc };
  if (sanitized._id) {
    sanitized._id = sanitized._id.toString();
  }
  return sanitized;
}

async function ingestRecord(record) {
  const response = await fetch(`${CERTYO_BASE_URL}/api/v1/records`, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      "X-API-Key": CERTYO_API_KEY,
    },
    body: JSON.stringify(record),
  });

  if (!response.ok) {
    const body = await response.text();
    console.error(`Ingestion failed for ${record.recordId}: ${response.status} ${body}`);
    return null;
  }

  return response.json();
}

main().catch((err) => {
  console.error("Fatal error:", err);
  process.exit(1);
});

"""
MongoDB polling worker for Certyo.
Queries documents modified since the last sync timestamp and ingests them.

Install: pip install pymongo requests
Usage:   CERTYO_API_KEY=xxx CERTYO_TENANT_ID=yyy python certyo_mongo_poll.py
"""

import os
import time
import json
import requests
from datetime import datetime, timezone
from pymongo import MongoClient
from bson import ObjectId

MONGO_URI = os.environ.get("MONGO_URI", "mongodb://localhost:27017/?replicaSet=rs0")
DATABASE = os.environ.get("MONGO_DATABASE", "productdb")
COLLECTION = os.environ.get("MONGO_COLLECTION", "products")

CERTYO_API_KEY = os.environ["CERTYO_API_KEY"]
CERTYO_TENANT_ID = os.environ["CERTYO_TENANT_ID"]
CERTYO_BASE_URL = os.environ.get("CERTYO_BASE_URL", "https://www.certyos.com")

POLL_INTERVAL = int(os.environ.get("POLL_INTERVAL", "30"))
BULK_THRESHOLD = 10


def get_last_sync(db):
    """Load the last sync timestamp from the _certyo_cursors collection."""
    cursor_doc = db["_certyo_cursors"].find_one({"_id": f"poll_{COLLECTION}"})
    if cursor_doc and "lastSyncAt" in cursor_doc:
        return cursor_doc["lastSyncAt"]
    return datetime(2000, 1, 1, tzinfo=timezone.utc)


def save_last_sync(db, sync_time):
    """Persist the sync timestamp."""
    db["_certyo_cursors"].update_one(
        {"_id": f"poll_{COLLECTION}"},
        {
            "$set": {
                "lastSyncAt": sync_time,
                "updatedAt": datetime.now(timezone.utc),
            }
        },
        upsert=True,
    )


def map_document(doc):
    """Map a MongoDB document to a Certyo record payload."""
    doc_id = str(doc["_id"])
    # Convert ObjectId and datetime fields for JSON serialization
    payload = {}
    for key, value in doc.items():
        if isinstance(value, ObjectId):
            payload[key] = str(value)
        elif isinstance(value, datetime):
            payload[key] = value.isoformat()
        else:
            payload[key] = value

    return {
        "tenantId": CERTYO_TENANT_ID,
        "database": DATABASE,
        "collection": COLLECTION,
        "recordId": doc_id,
        "recordVersion": "1",
        "operationType": "upsert",
        "recordPayload": payload,
        "sourceTimestamp": doc.get("updatedAt", datetime.now(timezone.utc)).isoformat()
            if isinstance(doc.get("updatedAt"), datetime)
            else datetime.now(timezone.utc).isoformat(),
        "idempotencyKey": f"mongo-poll-{COLLECTION}-{doc_id}-"
            f"{datetime.now(timezone.utc).strftime('%Y-%m-%d')}",
    }


def ingest_records(records):
    """Send records to Certyo — bulk if above threshold."""
    headers = {
        "X-API-Key": CERTYO_API_KEY,
        "Content-Type": "application/json",
    }

    if len(records) > BULK_THRESHOLD:
        # Chunk into batches of 1000
        for i in range(0, len(records), 1000):
            batch = records[i : i + 1000]
            payload = {"tenantId": CERTYO_TENANT_ID, "records": batch}
            resp = requests.post(
                f"{CERTYO_BASE_URL}/api/v1/records/bulk",
                headers=headers,
                json=payload,
            )
            if resp.ok:
                print(f"  Bulk ingested {len(batch)} records")
            else:
                print(f"  Bulk failed: {resp.status_code} {resp.text}")
    else:
        for record in records:
            resp = requests.post(
                f"{CERTYO_BASE_URL}/api/v1/records",
                headers=headers,
                json=record,
            )
            if resp.ok:
                result = resp.json()
                print(f"  Ingested {record['recordId']}: {result['recordHash']}")
            else:
                print(f"  Failed {record['recordId']}: {resp.status_code}")


def poll_cycle():
    """Run one polling cycle."""
    client = MongoClient(MONGO_URI)
    db = client[DATABASE]
    collection = db[COLLECTION]

    last_sync = get_last_sync(db)
    now = datetime.now(timezone.utc)

    # Query documents modified since last sync
    query = {"updatedAt": {"$gte": last_sync}}
    docs = list(collection.find(query).sort("updatedAt", 1).limit(5000))

    if not docs:
        client.close()
        return

    print(f"Found {len(docs)} modified documents since {last_sync.isoformat()}")
    records = [map_document(doc) for doc in docs]
    ingest_records(records)

    save_last_sync(db, now)
    client.close()


if __name__ == "__main__":
    print(f"MongoDB polling worker started (interval: {POLL_INTERVAL}s)")
    print(f"Watching {DATABASE}.{COLLECTION}")
    while True:
        try:
            poll_cycle()
        except Exception as e:
            print(f"Poll cycle error: {e}")
        time.sleep(POLL_INTERVAL)

version: "3.9"
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.6.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  kafka:
    image: confluentinc/cp-kafka:7.6.0
    depends_on: [zookeeper]
    ports: ["9092:9092"]
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,EXTERNAL://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  debezium:
    image: debezium/connect:2.5
    depends_on: [kafka]
    ports: ["8083:8083"]
    environment:
      BOOTSTRAP_SERVERS: kafka:29092
      GROUP_ID: certyo-mongo-cdc
      CONFIG_STORAGE_TOPIC: _debezium_configs
      OFFSET_STORAGE_TOPIC: _debezium_offsets
      STATUS_STORAGE_TOPIC: _debezium_status
      CONFIG_STORAGE_REPLICATION_FACTOR: 1
      OFFSET_STORAGE_REPLICATION_FACTOR: 1
      STATUS_STORAGE_REPLICATION_FACTOR: 1

# After startup, register the MongoDB connector:
#
# curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
#   "name": "certyo-mongo-source",
#   "config": {
#     "connector.class": "io.debezium.connector.mongodb.MongoDbConnector",
#     "mongodb.connection.string": "mongodb://mongo:27017/?replicaSet=rs0",
#     "topic.prefix": "certyo",
#     "collection.include.list": "productdb.products",
#     "capture.mode": "change_streams_update_full",
#     "snapshot.mode": "initial"
#   }
# }'
#
# This creates a Kafka topic: certyo.productdb.products
# Write a consumer that reads from this topic and POSTs to Certyo /api/v1/records