// streams.ts
export const STREAMS = {
  MARKET_EVENTS: {
    name: "market.events",
    subjects: ["market.events.>"],
    retention: "limits",
    maxAge: 7 * 24 * 60 * 60 * 1_000_000_000, // 7 days (nanoseconds)
    maxBytes: 10 * 1024 * 1024 * 1024, // 10 GB
    storage: "file",
    replicas: 3,
  },
  TRADE_OPPORTUNITIES: {
    name: "trade.opportunities",
    subjects: ["trade.opportunities.>"],
    retention: "limits",
    maxAge: 24 * 60 * 60 * 1_000_000_000, // 1 day
    maxBytes: 5 * 1024 * 1024 * 1024, // 5 GB
    storage: "file",
    replicas: 3,
  },
  PLANNED_ORDERS: {
    name: "execution.orders",
    subjects: ["execution.orders.>"],
    retention: "limits",
    maxAge: 30 * 24 * 60 * 60 * 1_000_000_000, // 30 days
    maxBytes: 20 * 1024 * 1024 * 1024, // 20 GB
    storage: "file",
    replicas: 3,
  },
};

import { connect, JSONCodec } from "nats";

const nc = await connect({ servers: process.env.NATS_URL });
const js = nc.jetstream();
const codec = JSONCodec();

// Publish
await js.publish("market.events.raydium.pool_update", codec.encode(event));

// Subscribe
const consumer = await js.consumers.get("market.events", "scanner-1");
const messages = await consumer.consume();

for await (const msg of messages) {
  const data = codec.decode(msg.data);
  await processEvent(data);
  msg.ack();
}

import Redis from "ioredis";

const redis = new Redis({
  host: process.env.REDIS_HOST,
  port: parseInt(process.env.REDIS_PORT),
  password: process.env.REDIS_PASSWORD,
  db: 0,
  retryStrategy: (times) => Math.min(times * 50, 2000),
});

// Set with TTL
await redis.setex(`price:${token}`, 300, JSON.stringify(priceData));

// Hash operations
await redis.hset("wallet:balances", address, balance.toString());

// Distributed lock
const lock = await redis.set(
  `lock:executor:${walletId}`,
  txId,
  "EX", 30,
  "NX"
);

import { Pool } from "pg";

const pool = new Pool({
  host: process.env.DB_HOST,
  port: parseInt(process.env.DB_PORT),
  database: process.env.DB_NAME,
  user: process.env.DB_USER,
  password: process.env.DB_PASSWORD,
  max: 20,
  idleTimeoutMillis: 30000,
  connectionTimeoutMillis: 2000,
});

// Transaction example
const client = await pool.connect();
try {
  await client.query("BEGIN");

  const tradeResult = await client.query(
    "INSERT INTO trades (strategy, wallet_id, signature, expected_profit, status) VALUES ($1, $2, $3, $4, $5) RETURNING id",
    [strategy, walletId, signature, expectedProfit, "pending"]
  );

  for (const route of routes) {
    await client.query(
      "INSERT INTO trade_routes (trade_id, step, protocol, pool, input_mint, output_mint, input_amount, output_amount) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)",
      [tradeResult.rows[0].id, route.step, route.protocol, route.pool, route.inputMint, route.outputMint, route.inputAmount, route.outputAmount]
    );
  }

  await client.query("COMMIT");
} catch (e) {
  await client.query("ROLLBACK");
  throw e;
} finally {
  client.release();
}

import { register, Counter, Histogram, Gauge } from "prom-client";

// Counters
export const opportunitiesDetected = new Counter({
  name: "opportunities_detected_total",
  help: "Total opportunities detected",
  labelNames: ["strategy"],
});

export const tradesExecuted = new Counter({
  name: "trades_executed_total",
  help: "Total trades executed",
  labelNames: ["strategy", "status"],
});

// Histograms
export const quoteLatency = new Histogram({
  name: "quote_latency_seconds",
  help: "Quote latency in seconds",
  labelNames: ["source"],
  buckets: [0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1],
});

// Gauges
export const activeWallets = new Gauge({
  name: "active_wallets",
  help: "Number of active wallets",
  labelNames: ["tier"],
});

// Usage
opportunitiesDetected.inc({ strategy: "arbitrage" });
quoteLatency.observe({ source: "solroute" }, 0.008);

import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node";
import { JaegerExporter } from "@opentelemetry/exporter-jaeger";
import { registerInstrumentations } from "@opentelemetry/instrumentation";
import { HttpInstrumentation } from "@opentelemetry/instrumentation-http";

const provider = new NodeTracerProvider();

const exporter = new JaegerExporter({
  endpoint: process.env.JAEGER_ENDPOINT,
});

provider.addSpanProcessor(
  new BatchSpanProcessor(exporter)
);

provider.register();

registerInstrumentations({
  instrumentations: [new HttpInstrumentation()],
});

// Usage
import { trace } from "@opentelemetry/api";

const tracer = trace.getTracer("arbitrage-planner");

async function analyzeOpportunity() {
  const span = tracer.startSpan("analyze_opportunity");

  try {
    const quote1 = await getQuote(); // auto-traced
    const quote2 = await getQuote(); // auto-traced

    span.addEvent("quotes_received");

    const profit = calculateProfit(quote1, quote2);
    span.setAttribute("profit", profit);

    return profit;
  } finally {
    span.end();
  }
}

import pino from "pino";

const logger = pino({
  level: process.env.LOG_LEVEL || "info",
  transport: {
    target: "pino-loki",
    options: {
      batching: true,
      interval: 5,
      host: process.env.LOKI_URL,
      labels: {
        app: "trading-system",
        service: "arbitrage-planner",
      },
    },
  },
});

logger.info({ opportunity: {...} }, "Opportunity detected");
logger.error({ err, txId }, "Transaction failed");