import { metricsPlugin, heartbeatPlugin, circuitBreakerPlugin } from "ocpp-ws-io/plugins";

server.plugin(
  metricsPlugin(),
  heartbeatPlugin(),
  circuitBreakerPlugin(),
);

┌─────────────────────────────────────────────────────────┐
│  Level 4 — Middleware           (mutates payloads)       │
│  pii-redactor · schema-versioning                        │
├─────────────────────────────────────────────────────────┤
│  Level 3 — Interceptor          (can drop messages)      │
│  message-dedup · replay-buffer                           │
├─────────────────────────────────────────────────────────┤
│  Level 2 — Lifecycle Controller  (acts on clients)       │
│  connection-guard · anomaly · circuit-breaker            │
├─────────────────────────────────────────────────────────┤
│  Level 1 — Passive Hook         (observes, never blocks) │
│  kafka · amqp · mqtt · redis-pubsub · webhook · metrics  │
│  otel · session-log · heartbeat · rate-limit-notifier    │
└─────────────────────────────────────────────────────────┘

server.plugin(
  // L4: Middleware — transform payloads before anything else sees them
  piiRedactorPlugin({ sensitiveKeys: ["idToken", "email"] }),
  schemaVersioningPlugin({ sourceVersion: "ocpp1.6", targetVersion: "ocpp2.0.1", rules }),

  // L3: Interceptor — deduplicate or replay before handlers process
  messageDedupPlugin({ redis }),
  replayBufferPlugin({ redis }),

  // L2: Lifecycle — guard and monitor connections
  connectionGuardPlugin({ maxConnections: 50_000 }),
  anomalyPlugin({ reconnectThreshold: 10 }),
  circuitBreakerPlugin({ failureThreshold: 5 }),

  // L1: Passive — observe and report
  metricsPlugin(),
  otelPlugin({ serviceName: "my-csms" }),
  heartbeatPlugin(),
  sessionLogPlugin(),
  webhookPlugin({ url: "https://api.example.com/events" }),
  kafkaPlugin({ producer }),
  rateLimitNotifierPlugin({ sink: "https://alerts.example.com" }),
);

import { piiRedactorPlugin } from "ocpp-ws-io/plugins";

server.plugin(piiRedactorPlugin({
  sensitiveKeys: ["password", "authorizationKey", "email", "phoneNumber"],
  replacement: "***",
}));

import { schemaVersioningPlugin } from "ocpp-ws-io/plugins";

server.plugin(schemaVersioningPlugin({
  sourceVersion: "ocpp1.6",
  targetVersion: "ocpp2.0.1",
  applyWhen: "ocpp1.6", // Only transform 1.6 clients
  rules: [
    {
      method: "BootNotification",
      transform: (payload, direction) => {
        if (direction === "up") {
          // 1.6 → 2.0.1: Wrap flat fields into chargingStation object
          return {
            chargingStation: {
              model: payload.chargePointModel,
              vendorName: payload.chargePointVendor,
              serialNumber: payload.chargePointSerialNumber,
              firmwareVersion: payload.firmwareVersion,
            },
            reason: "PowerUp",
          };
        }
        // 2.0.1 → 1.6: Flatten chargingStation back
        const cs = (payload.chargingStation ?? {}) as Record<string, unknown>;
        return {
          chargePointModel: cs.model,
          chargePointVendor: cs.vendorName,
          chargePointSerialNumber: cs.serialNumber,
          firmwareVersion: cs.firmwareVersion,
        };
      },
    },
    {
      method: "StatusNotification",
      transform: (payload, direction) => {
        if (direction === "up") {
          return {
            timestamp: new Date().toISOString(),
            connectorId: payload.connectorId,
            connectorStatus: payload.status === "Available" ? "Available" : "Occupied",
            evseId: payload.connectorId,
          };
        }
        return payload;
      },
    },
  ],
}));

Station (1.6) ──→ [Transform UP] ──→ Application Handler (2.0.1)
Station (1.6) ←── [Transform DOWN] ←── Application Handler (2.0.1)

import { messageDedupPlugin } from "ocpp-ws-io/plugins";
import Redis from "ioredis";

const redis = new Redis(process.env.REDIS_URL);

// With ioredis (default)
server.plugin(messageDedupPlugin({ redis }));

// With node-redis v4
import { createClient } from "redis";
const nodeRedis = createClient({ url: process.env.REDIS_URL });
await nodeRedis.connect();

server.plugin(messageDedupPlugin({
  redis: nodeRedis,
  redisStyle: "options",
  ttlMs: 60_000,
}));

import { replayBufferPlugin } from "ocpp-ws-io/plugins";

server.plugin(replayBufferPlugin({
  redis,
  flushConcurrency: 10,
  flushDelayMs: 100,
}));

// Queue a command for a disconnected station
await redis.rpush(
  "ocpp:replay:CP-001",
  JSON.stringify([2, "queued-1", "UnlockConnector", { connectorId: 1 }]),
);

// When CP-001 reconnects, the plugin automatically calls:
//   client.call("UnlockConnector", { connectorId: 1 })

import { connectionGuardPlugin } from "ocpp-ws-io/plugins";

server.plugin(connectionGuardPlugin({ maxConnections: 50_000 }));

import { anomalyPlugin } from "ocpp-ws-io/plugins";

server.plugin(anomalyPlugin({ reconnectThreshold: 10 }));

server.on("securityEvent", (evt) => {
  if (evt.type === "ANOMALY_RAPID_RECONNECT") {
    console.warn(`Reconnect storm: ${evt.identity}`);
  }
});

CLOSED ─(failures ≥ threshold)─→ OPEN
OPEN ─(resetTimeout expires)──→ HALF_OPEN
HALF_OPEN ─(success)───────────→ CLOSED
HALF_OPEN ─(failure)───────────→ OPEN

import { circuitBreakerPlugin } from "ocpp-ws-io/plugins";

server.plugin(circuitBreakerPlugin({
  failureThreshold: 3,
  resetTimeoutMs: 15_000,
  maxConcurrent: 10,
  onStateChange: (identity, from, to) => {
    console.log(`Circuit ${identity}: ${from} → ${to}`);
    if (to === "OPEN") {
      // Alert ops team — this station is flapping
      alerting.fire({ station: identity, event: "circuit_open" });
    }
  },
}));

import { heartbeatPlugin } from "ocpp-ws-io/plugins";

server.plugin(heartbeatPlugin());

import { metricsPlugin } from "ocpp-ws-io/plugins";

const metrics = metricsPlugin({
  intervalMs: 10_000,
  onSnapshot: (snap) => console.log(`Active: ${snap.activeConnections}`),
});
server.plugin(metrics);

// On-demand access
const snap = metrics.getMetrics();
// { activeConnections, peakConnections, totalConnections, totalValidationFailures, totalSecurityEvents, ... }

// Prometheus export
const lines = await metrics.getCustomMetrics();
// ["ocpp_connections_active 42", "ocpp_connections_peak 128", ...]

import { sessionLogPlugin } from "ocpp-ws-io/plugins";

server.plugin(sessionLogPlugin());
// => [session] connected { identity: "CP-101", ip: "192.168.1.1", protocol: "ocpp1.6" }
// => [session] disconnected { identity: "CP-101", durationSec: 3600, code: 1000 }

import { otelPlugin } from "ocpp-ws-io/plugins";

server.plugin(otelPlugin({ serviceName: "my-csms" }));

import { webhookPlugin } from "ocpp-ws-io/plugins";

server.plugin(webhookPlugin({
  url: "https://api.example.com/ocpp-events",
  secret: process.env.WEBHOOK_SECRET,
  events: ["connect", "disconnect"],
  headers: { Authorization: "Bearer token123" },
}));

import { kafkaPlugin } from "ocpp-ws-io/plugins";
import { Kafka } from "kafkajs";

const kafka = new Kafka({ brokers: ["kafka:9092"] });
const producer = kafka.producer();
await producer.connect();

server.plugin(kafkaPlugin({
  producer,
  topic: "ocpp.events",
  topicRouting: true, // → ocpp.events.connect, ocpp.events.message, etc.
  events: ["connect", "disconnect", "message", "security"],
}));

import { rateLimitNotifierPlugin } from "ocpp-ws-io/plugins";

// Simple webhook
server.plugin(rateLimitNotifierPlugin({
  sink: "https://slack.example.com/webhook",
  cooldownMs: 120_000,
  threshold: 3,
}));

// Custom sink (e.g., Kafka, PagerDuty, etc.)
server.plugin(rateLimitNotifierPlugin({
  sink: {
    send: async (alert) => {
      await pagerduty.trigger({
        summary: `Rate limit: ${alert.identity} (${alert.count} events)`,
        severity: "warning",
      });
    },
  },
  threshold: 5,
}));

import { createPlugin } from "ocpp-ws-io";

const myPlugin = createPlugin({
  name: "my-advanced-plugin",

  onInit(server) {
    console.log(`Plugin loaded on server listening at ${server.options.port}`);
  },

  onSecurityEvent(evt) {
    if (evt.type === "RATE_LIMIT_EXCEEDED") {
      console.warn(`[Security] Rate limit hit by ${evt.identity}`);
    }
  },

  onBeforeReceive(client, rawData) {
    // Arbitrary size limits or binary magic number checks
    if (rawData.length > 1024 * 1024) {
      console.error(`Payload too large from ${client.identity}`);
      return false; // Drops the message
    }
  },

  onMessage(client, payload) {
    // Unified observability: both outgoing and incoming messages
    const { message, direction, ctx } = payload;
    console.log(`[${direction}] ${ctx.action || ctx.type} at ${ctx.timestamp.toISOString()}`);
  },

  onTelemetry(stats) {
    // Push stats to your external monitoring system
    console.log(`Active Connections: ${stats.activeConnections}`);
  }
});

server.plugin(myPlugin);