support for nats.io as trigger?

[skurfuerst]

Hey,

would it be possible to support nats.io as trigger for windmill? I could use https://github.com/sandstorm/caddy-nats-bridge as a workaround, but I'd prefer direct integration to keep complexity low :)

All the best and keep up the great work,
Sebastian
PS: Sorry for not continuing my PR a long time ago - life got in the way ;)

[skurfuerst]

... thinking about it, even more powerful would be IMHO some kind of "Daemon Mode" script - i.e. a script which is running permanently (and restarted if it crashes). Then, this script could listen to NATS topics and trigger windmill logic as needed.

Curious about your thoughts :)

[rubenfiszel]

I think adding a script as a daemon is somewhat better suited to a sidecar so that it can be handled by docker itself. I do have the plan to standardize all such external queue and event system under a collection of "adapters". I wrote one for postgresql listen/notify in https://github.com/windmill-labs/windmill-adapters and I think nats could work the same way. We would have a side container or deno run script that would listen to nats events and forward to windmill. WDYT?

[skurfuerst]

Hey Ruben,

thanks for your thoughtful answer!

I am not 100% sure yet whether putting a sidecar outside windmill will actually help much for my use case (at least reducing complexity compared to what we have now). Let me explain:

• we use Nats.io as "nervous system"; so many services publish data to it at defined endpoints (in case of pub-sub); or accept requests (in case of request/response).
• A common case is the following: "When a certain message is published, trigger logic which does some requests to fetch data or modify state".
• We need quite a few scripts (each 100 or so lines long) which does this kind of plumbing / orchestration.
  • I have attached such a real world example script below.
• This script has two parts:
  • 1. listen to a certain event on a given topic ("an order was placed")
  • 2. then do the actual logic ("create a voucher for the placed order")

Doing part 2 in Windmill would be really easy; and I would LOVE the web UI, the observability, etc etc for all that :) I could basically copy/paste the code below in.

Part 1 is a bit hard, because the sending system defines the topic names which we need to listen on.

=> so we could do part 1 outside of windmill; but it would be great to have the logic "Listen to topic X, then for each message, trigger action" somewhere configured closely to the actual script. If we needed to place this outside, we'd miss some benefits of windmill regarding accessibility for our developers and ease of use.

Compared to webhooks, the difference is that for webhooks, we usually tell the sending system: "Please call URL X"; while when working with PubSub systems like NATS, we usually say "the sender publishes to sender.some.stream and anybody interested can listen to it". So we'd somehow need to make this part configurable on the windmill side I feel.

So if we could somehow make the triggers configurable, that would be huge :)

Thanks for your great work! It's awesome to see how far windmill has come ❤️

Sebastian

Example script

• we use Pretix to sell conference tickets.
• For every sold conference ticket, a personalized voucher code should be created.
• So we listen on an order via a webhook (see pretix-prod.PUB.webhook.orderPlaced below)
• ... and then we trigger logic by doing requests etc to other systems.

WRITE_VOUCHERS=1 deno run --allow-read --allow-net --allow-env main.ts

import { nats } from "./deps.ts";
import { ListResponse, Order, Voucher } from "./types.ts";

const WRITE_VOUCHERS = Deno.env.get("WRITE_VOUCHERS");

const ordersTopic = `pretix-prod.SRV.pretixapi.v1.GET.organizers.neoscon.events.2023.orders`;
const vouchersTopic = `pretix-prod.SRV.pretixapi.v1.GET.organizers.neoscon.events.2023.vouchers`;
const newVoucherTopic = `pretix-prod.SRV.pretixapi.v1.POST.organizers.neoscon.events.2023.vouchers`;
const deleteVoucherTopic = (voucherId: number) => `pretix-prod.SRV.pretixapi.v1.DELETE.organizers.neoscon.events.2023.vouchers.${voucherId}`;

const c = nats.StringCodec();

const conn = await nats.connect({
  servers: Deno.env.get("NATS_URL"),
  authenticator: nats.credsAuthenticator(
    new TextEncoder().encode(await Deno.readTextFile(Deno.env.get("NATS_CREDS_FILE") || "./auth.creds"))
  ),
  inboxPrefix: Deno.env.get("NATS_INBOX_PREFIX")
});

///////////// ORDER PLACED
(async () => {
  for await (const m of conn.subscribe(
    "pretix-prod.PUB.webhook.orderPlaced"
  )) {
    const order = JSON.parse(c.decode(m.data));
    console.log("Received orderPlaced: ", order)
    /*
    {
      notification_id: 3314,
      organizer: "neoscon",
      event: "2023",
      code: "AAAV",
      action: "pretix.event.order.placed"
    } 
    */
    const vouchers = await loadAllVouchers();
    ensureOrderHasFittingVoucher(order.code, vouchers);
  }
})();

///////////// ORDER CANCELED
(async () => {
  for await (const m of conn.subscribe(
    "k3s2021.pretix-prod.PUB.webhook.orderCanceled"
  )) {
    const order = JSON.parse(c.decode(m.data));
    console.log("Received orderCanceled: ", order)
    
    ensureVoucherForOrderDeleted(order.code);
  }
})();

///////////// ORDER REACTIVATED
(async () => {
  for await (const m of conn.subscribe(
    "k3s2021.pretix-prod.PUB.webhook.orderReactivated"
  )) {
    const order = JSON.parse(c.decode(m.data));
    console.log("Received orderReactivated: ", order)
    
    const vouchers = await loadAllVouchers();
    ensureOrderHasFittingVoucher(order.code, vouchers);
  }
})();

///////////// ON STARTUP -> CREATE ALL ORDERS
(async () => {
  const vouchers = await loadAllVouchers();
  const orders = await loadAllOrders();

  for await (const order of orders) {
    if (order.status == "c") {
      ensureVoucherForOrderDeleted(order.code);
    } else {
      ensureOrderHasFittingVoucher(order.code, vouchers);
    }
  }
  console.log("All done.");
})();


///////////// HELPERS
const voucherTemplate: Partial<Voucher> = {
  allow_ignore_quota: false,
  item: 93, // Full Conference Ticket
  tag: 'BringAFriend' // for grouping
};

const headers = nats.headers();
headers.set("Content-Type", `application/json`);

function voucherCodeGen(orderCode: string) {
  return `FRIEND_${orderCode}`;
}

async function ensureOrderHasFittingVoucher(
  orderCode: string,
  vouchers: VouchersByCode
) {
  const voucherCode = voucherCodeGen(orderCode);

  if (vouchers[voucherCode]) {
    console.log(
      `Voucher ${voucherCode} already exists. Not modified.`
    );
  } else {
    console.log(`Voucher ${voucherCode} does not exist. Creating it.`);
    const voucher = Object.assign({}, voucherTemplate, {
      code: voucherCode,
      // Voucher Price is not needed; this happens through the new product.
    });

    if (WRITE_VOUCHERS) {
      const response = await conn.request(
        newVoucherTopic,
        c.encode(JSON.stringify(voucher)),
        { headers, timeout: 3000 }
      );
      console.log(voucherCode, c.decode(response.data));
    } else {
      console.log("Dry run");
    }
  }
}

async function ensureVoucherForOrderDeleted(orderCode: string) {
  const vouchers = await loadAllVouchers();
  const voucher = vouchers[voucherCodeGen(orderCode)];
  
  if (voucher?.id) {
    console.log(`Deleting voucher ${voucher.id} for order ${orderCode}`);
    conn.publish(deleteVoucherTopic(voucher.id));
  } else {
    console.warn(`voucher for order ${orderCode} not found - doing nothing (as it should not exist).`);
  }
}


async function* loadAllOrders(): AsyncGenerator<Order> {
  yield* loadAll(ordersTopic) as AsyncGenerator<Order>;
}

type VouchersByCode = { [code: string]: Voucher };

async function loadAllVouchers(): Promise<VouchersByCode> {
  const vouchers = loadAll(vouchersTopic) as AsyncGenerator<Voucher>;
  const result: VouchersByCode = {};
  for await (const voucher of vouchers) {
    result[voucher.code] = voucher;
  }

  return result;
}

/**
 * Helper to load all elements of a Pretix Listing lazily
 *
 * @param topic
 * @returns
 */
async function* loadAll(topic: string): AsyncGenerator<unknown, undefined> {
  let page = 1;
  while (1) {
    const h = nats.headers();
    h.set("X-NatsBridge-UrlQuery", `page=${page}`);

    const res = await conn.request(topic, undefined, {
      headers: h,
      timeout: 3000,
    });

    const str = c.decode(res.data);
    const response = JSON.parse(str) as ListResponse<unknown>;

    for (const element of response.results) {
      yield element;
    }
    if (response.next === null) {
      // ended at last page, finished
      return;
    }

    page++;
  }
}

[rubenfiszel]

That makes a lot of sense but I think to reduce complexity you could still make it work using a sidecar AND use autodiscovery.
For instance, you could have a naming scheme such that all scripts and flows under f/nats_discovery/mychannel will listen to mychannel.
This way, you can natively integrate into windmill current scheme without modifications and all the 'complexity' would be inside the sidecar.

It's not as complex as one might think because windmill expose webhooks that are triggered upon any changes such as deploying scripts and flows.

I am also open to have a native integration of NATS in windmill but that is a lot more complexity in both backend and frontend and I would think that one would want to a v0 of such system before investing too much in that direction.

The sidecar would behave as follows.

Maintain a table:
channel -> Array<windmill endpoints>

At initialization, browse windmill scripts/flows and populate the table ^

Keep an HTTP server open to listen to changes from the configured change webhook
    - on any change, modify table above

Listen to all NATS channels (or all channels whitelisted or passed as args)
    - for each message, send the corresponding webhook given the table above

WDYT ?

[skurfuerst]

Hey @rubenfiszel ,

Good idea, I will try this out (maybe with some modification; so that I include a script comment listing the to-be-listened-on NATS channels - so one can listen to multiple channels and wildcards etc). I will report back once I tried it :)

I am not sure whether I would recommend a "native" NATS integration for you; because the next one would suggest rabbitMQ, MQTT, or any other protocol ;)

Thanks, keep up the great work and all the best,
Sebastian

[skurfuerst]

Hey @rubenfiszel ,

short update: the system basically works, I will do some cleanups and then post a working example in the upcoming days 😊

One problem I still see is that there is no feedback to the user in case he made an error etc - for example it can happen that the Nats user has no permissions to access a certain topic.

One maybe pragmatic way to solve this would be to allow some kind of external "Validation" webhook/script/... which gets the script text as input before executing it, when the user tests it through the UI, and which can return errors or success messages displayed to the user inside the UI.

I'd be glad to hear your thoughts on this idea :)

All the best,
And keep up the great work,
Sebastian

[rubenfiszel]

One idea would probably to have a cron job that is regularly scheduled to check for mismatch in permission and send an alert to the user if it is detected. You can also listen to the instance webhook for all script/flow save events instead of cron

[rubenfiszel]

@skurfuerst still happy about the system ? :)

[skurfuerst]

@rubenfiszel yeah :) working really nicely so far, though I am still gaining experience. Right now I am for a month on parental leave traveling; so my time on the laptop is quite limited :) that's why I also did not publish anything yet.

All the best,
Sebastian