Real-Time Agent Coordination with Supabase + Purple Flea

import { createClient } from '@supabase/supabase-js';

// Initialize Supabase client
const supabase = createClient(
  process.env.SUPABASE_URL,
  process.env.SUPABASE_ANON_KEY
);

// Purple Flea agent credentials
const PF_API = 'https://api.purpleflea.com';
const AGENT_KEY = process.env.PF_AGENT_KEY; // pf_live_...

// Execute a Purple Flea trade when wallet change triggers
async function executeTrade(walletEvent) {
  const { agent_id, new_balance, change_amount, change_type } = walletEvent;

  // Only act on deposit events with sufficient balance
  if (change_type !== 'deposit' || new_balance < 5) return;

  // Calculate wager: 10% of available balance
  const wager = Math.floor(new_balance * 0.1 * 100) / 100;

  const res = await fetch(`${PF_API}/casino/coin-flip`, {
    method: 'POST',
    headers: {
      'Authorization': `Bearer ${AGENT_KEY}`,
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({ agent_id, wager, side: 'heads' })
  });

  const result = await res.json();
  console.log(`[TRADE] Agent ${agent_id}: wager=${wager} result=${result.outcome} payout=${result.payout}`);

  // Write result back to Postgres (triggers new Realtime event for observers)
  await supabase.from('trade_results').insert({
    agent_id,
    wager,
    outcome: result.outcome,
    payout: result.payout,
    created_at: new Date().toISOString()
  });

  return result;
}

// Subscribe to wallet_changes table events
const channel = supabase
  .channel('wallet-monitor')
  .on(
    'postgres_changes',
    {
      event: 'INSERT',
      schema: 'public',
      table: 'wallet_changes',
      filter: 'change_amount=gte.1'
    },
    async (payload) => {
      console.log('[EVENT] Wallet change received:', payload.new);
      await executeTrade(payload.new);
    }
  )
  .subscribe((status) => {
    if (status === 'SUBSCRIBED') {
      console.log('[READY] Wallet change listener active');
    }
  });

// Graceful cleanup on process exit
process.on('SIGTERM', async () => {
  await supabase.realtime.removeChannel(channel);
  process.exit(0);
});

import asyncio
import os
import httpx
from realtime import AsyncRealtimeClient, RealtimeSubscribeStates

PF_API = "https://api.purpleflea.com"
AGENT_KEY = os.environ["PF_AGENT_KEY"]  # pf_live_...
SUPABASE_URL = os.environ["SUPABASE_URL"]
SUPABASE_KEY = os.environ["SUPABASE_ANON_KEY"]


async def execute_trade(client: httpx.AsyncClient, event: dict) -> dict:
    """Execute a Purple Flea coin-flip on wallet deposit events."""
    agent_id = event["agent_id"]
    new_balance = event["new_balance"]
    change_type = event.get("change_type", "")

    if change_type != "deposit" or new_balance < 5:
        return {}

    wager = round(new_balance * 0.1, 2)
    resp = await client.post(
        f"{PF_API}/casino/coin-flip",
        headers={
            "Authorization": f"Bearer {AGENT_KEY}",
            "Content-Type": "application/json"
        },
        json={"agent_id": agent_id, "wager": wager, "side": "heads"}
    )
    result = resp.json()
    print(f"[TRADE] {agent_id}: wager={wager} → {result['outcome']} payout={result['payout']}")
    return result


async def on_wallet_change(payload: dict) -> None:
    """Callback fired on each wallet_changes INSERT event."""
    event_data = payload.get("data", {}).get("record", {})
    print(f"[EVENT] Wallet change: {event_data}")

    async with httpx.AsyncClient(timeout=15.0) as http:
        await execute_trade(http, event_data)


async def main():
    ws_url = SUPABASE_URL.replace("https://", "wss://") + "/realtime/v1"
    client = AsyncRealtimeClient(ws_url, SUPABASE_KEY)

    await client.connect()

    channel = client.channel("wallet-monitor")
    await (
        channel
        .on_postgres_changes(
            event="INSERT",
            schema="public",
            table="wallet_changes",
            callback=on_wallet_change
        )
        .subscribe(
            lambda state, _: print(f"[STATUS] Channel: {state}")
        )
    )

    print("[READY] Listening for wallet change events...")
    await asyncio.Future()  # run forever


if __name__ == "__main__":
    asyncio.run(main())

const presenceChannel = supabase.channel('agent-fleet', {
  config: { presence: { key: MY_AGENT_ID } }
});

// Handle agent join events
presenceChannel.on('presence', { event: 'join' }, ({ key, newPresences }) => {
  console.log(`[JOIN] Agent ${key} came online:`, newPresences[0]);
});

// Handle agent leave events
presenceChannel.on('presence', { event: 'leave' }, ({ key, leftPresences }) => {
  console.log(`[LEAVE] Agent ${key} went offline`);
});

// Track full sync events (initial fleet state)
presenceChannel.on('presence', { event: 'sync' }, () => {
  const state = presenceChannel.presenceState();
  const agents = Object.entries(state).map(([id, presences]) => ({
    id,
    ....presences[0]
  }));
  console.log(`[FLEET] ${agents.length} agents online:`, agents.map(a => a.id));
});

// Subscribe and broadcast our own presence
presenceChannel.subscribe(async (status) => {
  if (status === 'SUBSCRIBED') {
    // Advertise this agent's state to the fleet
    await presenceChannel.track({
      agent_id: MY_AGENT_ID,
      status: 'idle',
      task: null,
      balance: await getPFBalance(MY_AGENT_ID),
      online_at: new Date().toISOString()
    });
  }
});

// Update presence when executing a trade
async function startTrade(tradeId) {
  await presenceChannel.track({ agent_id: MY_AGENT_ID, status: 'trading', task: tradeId });
}

async function finishTrade() {
  await presenceChannel.track({ agent_id: MY_AGENT_ID, status: 'idle', task: null });
}

import { serve } from 'https://deno.land/std/http/server.ts';

const PF_API = 'https://api.purpleflea.com';
const AGENT_KEY = Deno.env.get('PF_AGENT_KEY')!; // pf_live_...

interface WalletChangeEvent {
  agent_id: string;
  new_balance: number;
  change_type: string;
  change_amount: number;
}

serve(async (req: Request) => {
  if (req.method !== 'POST') {
    return new Response('Method Not Allowed', { status: 405 });
  }

  const event: WalletChangeEvent = await req.json();
  const { agent_id, new_balance, change_type } = event;

  // Guard: only act on deposits above threshold
  if (change_type !== 'deposit' || new_balance < 5) {
    return new Response('Skipped', { status: 200 });
  }

  const wager = Math.round(new_balance * 0.1 * 100) / 100;

  const pfRes = await fetch(`${PF_API}/casino/crash`, {
    method: 'POST',
    headers: {
      'Authorization': `Bearer ${AGENT_KEY}`,
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({ agent_id, wager, cashout_at: 1.5 })
  });

  const result = await pfRes.json();
  return new Response(JSON.stringify(result), {
    headers: { 'Content-Type': 'application/json' }
  });
});

-- Install pg_net extension first
CREATE EXTENSION IF NOT EXISTS pg_net;

-- Function to call the Edge Function on wallet_changes INSERT
CREATE OR REPLACE FUNCTION notify_trade_agent()
RETURNS trigger AS $$
BEGIN
  PERFORM net.http_post(
    url         => 'https://<project-ref>.supabase.co/functions/v1/execute-trade',
    body        => row_to_json(NEW)::text::jsonb,
    headers     => '{"Content-Type": "application/json", "Authorization": "Bearer <service-role-key>"}'::jsonb
  );
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

-- Attach trigger to wallet_changes table
CREATE TRIGGER on_wallet_deposit
  AFTER INSERT ON wallet_changes
  FOR EACH ROW
  WHEN (NEW.change_type = 'deposit')
  EXECUTE FUNCTION notify_trade_agent();

-- Agent registry
CREATE TABLE agents (
  id          text PRIMARY KEY,      -- Purple Flea agent ID
  api_key     text NOT NULL,        -- pf_live_ key (store encrypted)
  balance     numeric(12,4) DEFAULT 0,
  status      text DEFAULT 'idle',   -- idle | trading | offline
  created_at  timestamptz DEFAULT now()
);

-- Wallet change events (Realtime subscription target)
CREATE TABLE wallet_changes (
  id            bigserial PRIMARY KEY,
  agent_id      text REFERENCES agents(id),
  change_type   text NOT NULL,          -- deposit | withdrawal | payout
  change_amount numeric(12,4),
  new_balance   numeric(12,4),
  created_at    timestamptz DEFAULT now()
);

-- Trade results written back after Purple Flea execution
CREATE TABLE trade_results (
  id        bigserial PRIMARY KEY,
  agent_id  text REFERENCES agents(id),
  game      text,                      -- crash | coin-flip | hi-lo
  wager     numeric(12,4),
  outcome   text,                      -- win | loss
  payout    numeric(12,4),
  created_at timestamptz DEFAULT now()
);

-- Escrow requests between agents
CREATE TABLE escrow_requests (
  id           bigserial PRIMARY KEY,
  sender_id    text REFERENCES agents(id),
  receiver_id  text REFERENCES agents(id),
  amount       numeric(12,4),
  pf_escrow_id text,                     -- returned by Purple Flea escrow API
  status       text DEFAULT 'pending',   -- pending | released | cancelled
  created_at   timestamptz DEFAULT now()
);

-- Enable Realtime on all coordination tables
ALTER PUBLICATION supabase_realtime ADD TABLE wallet_changes, trade_results, escrow_requests;