Callisto Iris

Database-agnostic event orchestration. Zero brokers.

Iris is a database-agnostic event orchestration engine, with typed routing, auto-reconnect, retry logic, and dead-letter handling, built without Kafka, RabbitMQ, or any external message broker. The core, transport, detection, routing, retry, is all interface-driven; Postgres LISTEN/NOTIFY is the first specialization built on top of it.

---

Why Iris

If you're already on a database, you don't need a separate broker for internal event-driven workflows. Iris's core is the Java engine and algorithm, not any one database binding; database-specific behavior lives behind small interfaces, and each database gets its own specialization implementing them.

• Zero new infrastructure — if you have a supported database, you have the whole stack
• Database-agnostic core — transport and change detection are both pluggable interfaces; Postgres is the first implementation of each
• Typed routing — each channel routes to its own handler
• Retry + dead-letter — failed handlers retry with backoff; unrecoverable events land in a dead-letter channel
• Auto-reconnect (Postgres specialization) — survives restarts and host reboots; re-registers all channels automatically

---

Quickstart (Postgres specialization)

1. Install the trigger function (once per database)

-- sql/iris_notify.sql
CREATE OR REPLACE FUNCTION iris_notify()
RETURNS trigger AS $$
BEGIN
    IF NEW.processed_at IS NULL THEN
        PERFORM pg_notify(
            TG_ARGV[0],
            row_to_json(NEW)::text
        );
    END IF;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

2. Attach to any table

CREATE TRIGGER orders_iris_trigger
    AFTER INSERT OR UPDATE ON orders
    FOR EACH ROW
    EXECUTE FUNCTION iris_notify('order_status_changed');

The trigger only fires when processed_at IS NULL — preventing cyclic re-triggering when your handler writes back to the same row.

3. Wire up the pipeline in Java

PgTransportConfig config = PgTransportConfig.builder()
    .host("localhost")
    .database("mydb")
    .username("postgres")
    .password("secret")
    .initialBackoffMs(1000)
    .maxBackoffMs(30000)
    .build();

IrisPipeline pipeline = IrisPipeline.builder(new PgTransport(config))
    .listen("order_status_changed", (channel, payload) -> {
        // parse payload, call your service, write back processed_at
    })
    .listen("inventory_updated", (channel, payload) -> {
        // handle inventory events
    })
    .maxRetries(3)
    .build();

pipeline.start();

---

Cycle Guard

The iris_notify() function uses processed_at IS NULL as a sentinel to prevent infinite trigger loops:

processed_at	Meaning
NULL	Pending — trigger will fire
timestamp	Processed — trigger will not re-fire

Your handler sets processed_at = NOW() on write-back. If the handler crashes before writing back, the row stays NULL and will be retried on the next event or a scheduled sweep.

---

Auto-Reconnect

Iris automatically detects connection drops via pgjdbc-ng's PGNotificationListener.closed() callback and reconnects with exponential backoff. All LISTEN subscriptions are re-registered on the new connection — no manual intervention required.

Connection drops (Postgres restart / host reboot)
    → closed() fires
    → exponential backoff reconnect loop
    → new connection established
    → all channels re-registered automatically

---

Transport Interface

Iris is DB-agnostic by design. The transport layer is a pluggable interface:

public interface IrisTransport {
    void connect();
    void subscribe(String channel, TransportListener listener);
    void unsubscribe(String channel);
    void disconnect();
    boolean isConnected();
}

PgTransport is the Postgres implementation. Additional transports can be added without changing routing, cycle guard, or handler logic.

---

Built by

Callisto Tech — tools for event-driven data infrastructure.