Your database is already a message stream. Give it judgment.

// npm i firebase-functions firebase-admin
const { onDocumentCreated } = require('firebase-functions/v2/firestore');
const { initializeApp } = require('firebase-admin/app');
initializeApp();

exports.validaitOnMessage = onDocumentCreated('rooms/{roomId}/messages/{msgId}', async (event) => {
  const snap = event.data;
  if (!snap) return;
  const res = await fetch('https://api.validait.dev/v1/events', {
    method: 'POST',
    headers: { 'content-type': 'application/json', authorization: `Bearer ${process.env.VALIDAIT_KEY}` },
    body: JSON.stringify({
      source: 'firestore',
      collection: 'messages',
      id: snap.id,
      fields: { text: snap.get('text'), authorId: snap.get('authorId') },
    }),
  });
  await snap.ref.update({ validait: await res.json() });
});

create extension if not exists pg_net with schema extensions;

create or replace function public.validait_on_message() returns trigger
language plpgsql security definer as $$
begin
  perform net.http_post(
    url := 'https://api.validait.dev/v1/events',
    headers := jsonb_build_object(
      'Content-Type', 'application/json',
      'Authorization', 'Bearer ' || current_setting('app.validait_key', true)
    ),
    body := jsonb_build_object(
      'source', 'supabase',
      'collection', tg_table_name,
      'id', new.id,
      'fields', jsonb_build_object('text', new.text, 'authorId', new.author_id)
    ),
    timeout_milliseconds := 5000
  );
  return new;
end;
$$;

create trigger validait_messages_ai after insert on public.messages
for each row execute function public.validait_on_message();

// Appwrite Function, trigger: tablesdb.*.tables.messages.rows.*.create
import { Client, TablesDB } from 'node-appwrite';

export default async ({ req, res }) => {
  const row = JSON.parse(req.bodyRaw);
  const r = await fetch('https://api.validait.dev/v1/events', {
    method: 'POST',
    headers: { 'content-type': 'application/json', authorization: `Bearer ${process.env.VALIDAIT_KEY}` },
    body: JSON.stringify({
      source: 'appwrite',
      collection: 'messages',
      id: row.$id,
      fields: { text: row.text, authorId: row.authorId },
    }),
  });
  const client = new Client()
    .setEndpoint(process.env.APPWRITE_FUNCTION_API_ENDPOINT)
    .setProject(process.env.APPWRITE_FUNCTION_PROJECT_ID)
    .setKey(req.headers['x-appwrite-key']);
  await new TablesDB(client).updateRow({
    databaseId: 'main', tableId: 'messages', rowId: row.$id,
    data: { validait: JSON.stringify(await r.json()) },
  });
  return res.empty();
};

// convex/messages.ts
import { mutation, internalAction } from './_generated/server';
import { internal } from './_generated/api';
import { v } from 'convex/values';

export const send = mutation({
  args: { text: v.string(), authorId: v.string() },
  handler: async (ctx, args) => {
    const id = await ctx.db.insert('messages', args);
    await ctx.scheduler.runAfter(0, internal.messages.check, { id });
  },
});

export const check = internalAction({
  args: { id: v.id('messages') },
  handler: async (ctx, { id }) => {
    const m = await ctx.runQuery(internal.messages.get, { id });
    const r = await fetch('https://api.validait.dev/v1/events', {
      method: 'POST',
      headers: { 'content-type': 'application/json', authorization: `Bearer ${process.env.VALIDAIT_KEY}` },
      body: JSON.stringify({
        source: 'convex',
        collection: 'messages',
        id,
        fields: { text: m.text, authorId: m.authorId },
      }),
    });
    await ctx.runMutation(internal.messages.save, { id, validait: await r.json() });
  },
});

exports = async function (changeEvent) {
  const doc = changeEvent.fullDocument;
  const r = await context.http.post({
    url: 'https://api.validait.dev/v1/events',
    headers: {
      'Content-Type': ['application/json'],
      Authorization: ['Bearer ' + context.values.get('VALIDAIT_KEY')],
    },
    body: JSON.stringify({
      source: 'mongodb',
      collection: changeEvent.ns.coll,
      id: doc._id.toString(),
      fields: { text: doc.text, authorId: doc.authorId },
    }),
  });
  const validait = EJSON.parse(r.body.text());
  await context.services
    .get('mongodb-atlas')
    .db(changeEvent.ns.db)
    .collection(changeEvent.ns.coll)
    .updateOne({ _id: doc._id }, { $set: { validait } });
};

// pb_hooks/validait.pb.js
onRecordAfterCreateSuccess((e) => {
  const res = $http.send({
    url: 'https://api.validait.dev/v1/events',
    method: 'POST',
    headers: { 'content-type': 'application/json', authorization: 'Bearer ' + $os.getenv('VALIDAIT_KEY') },
    body: JSON.stringify({
      source: 'pocketbase',
      collection: e.record.collection().name,
      id: e.record.id,
      fields: { text: e.record.getString('text'), authorId: e.record.getString('authorId') },
    }),
    timeout: 10,
  });
  e.record.set('validait', res.json);
  e.app.save(e.record);
  e.next();
}, 'messages');

curl -X POST https://api.validait.dev/v1/events \
  -H 'content-type: application/json' \
  -H "authorization: Bearer $VALIDAIT_KEY" \
  -d '{
        "source": "my-api",
        "collection": "messages",
        "id": "msg_123",
        "fields": {
          "text": "hey, lets finish this on whatsapp",
          "authorId": "u_42",
          "roomId": "r_7"
        }
      }'