bitvid/js/services/dmNostrService.js

import { ensureNostrTools, resolveSimplePoolConstructor, shimLegacySimplePoolMethods } from "../nostr/toolkit.js";
import { sanitizeRelayList } from "../nostr/nip46Client.js";
import { publishEventToRelay } from "../nostrPublish.js";
import logger from "../utils/logger.js";

const DEFAULT_BACKOFF = {
  baseMs: 1000,
  maxMs: 30000,
  jitterRatio: 0.3,
};

const MAX_SEEN_EVENT_IDS = 5000;
export const DM_RELAY_WARNING_FALLBACK = "dm-relays-fallback";

function normalizeRelayList(relays) {
  if (!Array.isArray(relays)) {
    return [];
  }

  return Array.from(
    new Set(
      relays
        .map((relay) => (typeof relay === "string" ? relay.trim() : ""))
        .filter(Boolean),
    ),
  );
}

function normalizePubkey(value) {
  if (typeof value !== "string") {
    return "";
  }
  return value.trim().toLowerCase();
}

function normalizeLogger(candidate) {
  if (
    candidate &&
    typeof candidate === "object" &&
    candidate.dev &&
    candidate.user
  ) {
    return candidate;
  }

  return logger;
}

function extractRelayHintsFromEvent(event) {
  const tags = Array.isArray(event?.tags) ? event.tags : [];
  return sanitizeRelayList(
    tags
      .filter((tag) => Array.isArray(tag) && tag[0] === "relay")
      .map((tag) => (typeof tag[1] === "string" ? tag[1].trim() : "")),
  );
}

export async function resolveDmRelaySelection({
  pubkey,
  relayHints = [],
  discoveryRelays = [],
  fallbackRelays = [],
  pool = null,
  log = null,
} = {}) {
  const resolvedLogger = normalizeLogger(log || logger);
  const normalizedPubkey = normalizePubkey(pubkey);
  const hintedRelays = sanitizeRelayList(Array.isArray(relayHints) ? relayHints : []);

  if (hintedRelays.length) {
    return { relays: hintedRelays, source: "hints", warning: null };
  }

  const discoveryList = sanitizeRelayList(
    Array.isArray(discoveryRelays) ? discoveryRelays : [],
  );
  const fallbackList = sanitizeRelayList(
    Array.isArray(fallbackRelays) && fallbackRelays.length
      ? fallbackRelays
      : discoveryRelays,
  );

  if (
    normalizedPubkey &&
    pool &&
    typeof pool.list === "function" &&
    discoveryList.length
  ) {
    try {
      const events = await pool.list(discoveryList, [
        { kinds: [10050], authors: [normalizedPubkey], limit: 1 },
      ]);
      const sorted = Array.isArray(events)
        ? events
            .filter((entry) => entry && entry.pubkey === normalizedPubkey)
            .sort((a, b) => (b?.created_at || 0) - (a?.created_at || 0))
        : [];
      if (sorted.length) {
        const relays = extractRelayHintsFromEvent(sorted[0]);
        if (relays.length) {
          return { relays, source: "kind-10050", warning: null };
        }
      }
    } catch (error) {
      resolvedLogger.dev.warn("[dmNostrService] Failed to fetch DM relay hints.", error);
    }
  }

  if (fallbackList.length) {
    return {
      relays: fallbackList,
      source: "fallback",
      warning: DM_RELAY_WARNING_FALLBACK,
    };
  }

  return { relays: [], source: "none", warning: null };
}

function clampNumber(value, minimum, maximum) {
  if (!Number.isFinite(value)) {
    return minimum;
  }

  return Math.min(Math.max(value, minimum), maximum);
}

function resolveBackoffConfig(custom = {}) {
  return {
    baseMs: clampNumber(custom.baseMs ?? DEFAULT_BACKOFF.baseMs, 250, 60000),
    maxMs: clampNumber(custom.maxMs ?? DEFAULT_BACKOFF.maxMs, 1000, 300000),
    jitterRatio: clampNumber(custom.jitterRatio ?? DEFAULT_BACKOFF.jitterRatio, 0, 1),
  };
}

function applyJitter(value, ratio) {
  if (!Number.isFinite(value) || value <= 0) {
    return 0;
  }
  if (!Number.isFinite(ratio) || ratio <= 0) {
    return value;
  }
  const jitter = value * ratio;
  return Math.round(value + (Math.random() * jitter - jitter / 2));
}

class DmNostrService {
  constructor({
    relays = [],
    preferredRelays = [],
    backoff = {},
    log = null,
  } = {}) {
    this.relays = normalizeRelayList(relays);
    this.preferredRelays = normalizeRelayList(preferredRelays);
    this.pool = null;
    this.poolPromise = null;
    this.logger = normalizeLogger(log || logger);
    this.backoffConfig = resolveBackoffConfig(backoff);
    this.relayStates = new Map();
    this.relayWatermarks = new Map();
    this.relayTimers = new Map();
    this.seenEventIds = new Set();
    this.seenEventOrder = [];
    this.subscriptions = new Map();
    this.activeFilters = [];
    this.actorPubkey = "";
    this.handlers = {
      onEvent: null,
      onEose: null,
      onError: null,
    };
  }

  async ensurePool() {
    if (this.pool) {
      return this.pool;
    }

    if (this.poolPromise) {
      return this.poolPromise;
    }

    this.poolPromise = (async () => {
      const tools = await ensureNostrTools();
      const SimplePool = resolveSimplePoolConstructor(tools);
      if (typeof SimplePool !== "function") {
        throw new Error("nostr-tools SimplePool unavailable");
      }

      const pool = new SimplePool();
      shimLegacySimplePoolMethods(pool);
      this.pool = pool;
      this.poolPromise = null;
      return pool;
    })();

    try {
      return await this.poolPromise;
    } catch (error) {
      this.poolPromise = null;
      this.logger.user.warn("[dmNostrService] Failed to initialize pool.", error);
      throw error;
    }
  }

  getRelayStatus(relayUrl) {
    const relay = this.relayStates.get(relayUrl);
    return relay ? { ...relay } : null;
  }

  listRelayStatuses() {
    return Array.from(this.relayStates.entries()).map(([url, state]) => ({
      url,
      ...state,
    }));
  }

  async connect({ relays = null } = {}) {
    if (Array.isArray(relays)) {
      this.relays = normalizeRelayList(relays);
    }

    if (!this.relays.length) {
      this.logger.user.warn("[dmNostrService] No relays configured for DM service.");
      return;
    }

    await this.ensurePool();

    await Promise.all(
      this.relays.map(async (relayUrl) => {
        await this.connectRelay(relayUrl);
      }),
    );
  }

  async connectRelay(relayUrl) {
    if (!this.pool) {
      await this.ensurePool();
    }

    if (!relayUrl) {
      return;
    }

    const relayState = this.ensureRelayState(relayUrl);
    relayState.lastConnectAttempt = Date.now();

    if (typeof this.pool.ensureRelay === "function") {
      try {
        const relay = await this.pool.ensureRelay(relayUrl);
        relayState.connected = Boolean(relay);
        if (relay) {
          this.attachRelayListeners(relayUrl, relay);
          this.recordRelaySuccess(relayUrl);
        }
      } catch (error) {
        this.recordRelayFailure(relayUrl, error);
        this.scheduleReconnect(relayUrl, "connect-failed");
      }
    } else {
      relayState.connected = true;
      this.recordRelaySuccess(relayUrl);
    }
  }

  attachRelayListeners(relayUrl, relay) {
    if (!relay || typeof relay.on !== "function") {
      return;
    }

    try {
      relay.on("connect", () => {
        this.recordRelaySuccess(relayUrl);
      });
      relay.on("disconnect", () => {
        this.recordRelayFailure(relayUrl, new Error("relay-disconnect"));
        this.scheduleReconnect(relayUrl, "relay-disconnect");
      });
      relay.on("error", (error) => {
        this.recordRelayFailure(relayUrl, error);
        this.scheduleReconnect(relayUrl, "relay-error");
      });
    } catch (error) {
      this.logger.dev.warn("[dmNostrService] Failed to attach relay listeners.", error);
    }
  }

  ensureRelayState(relayUrl) {
    if (!this.relayStates.has(relayUrl)) {
      this.relayStates.set(relayUrl, {
        connected: false,
        okCount: 0,
        failCount: 0,
        lastOkAt: 0,
        lastFailAt: 0,
        lastConnectAttempt: 0,
        backoffMs: 0,
        attempts: 0,
      });
    }

    return this.relayStates.get(relayUrl);
  }

  recordRelaySuccess(relayUrl) {
    const relayState = this.ensureRelayState(relayUrl);
    relayState.connected = true;
    relayState.okCount += 1;
    relayState.lastOkAt = Date.now();
    relayState.backoffMs = 0;
    relayState.attempts = 0;
  }

  recordRelayFailure(relayUrl, error) {
    const relayState = this.ensureRelayState(relayUrl);
    relayState.connected = false;
    relayState.failCount += 1;
    relayState.lastFailAt = Date.now();

    this.logger.dev.warn("[dmNostrService] Relay failure", {
      relayUrl,
      error,
    });
  }

  scheduleReconnect(relayUrl, reason) {
    const relayState = this.ensureRelayState(relayUrl);
    relayState.attempts += 1;

    const backoffBase = this.backoffConfig.baseMs * 2 ** (relayState.attempts - 1);
    const backoffMs = applyJitter(
      Math.min(backoffBase, this.backoffConfig.maxMs),
      this.backoffConfig.jitterRatio,
    );

    relayState.backoffMs = backoffMs;

    if (this.relayTimers.has(relayUrl)) {
      clearTimeout(this.relayTimers.get(relayUrl));
    }

    this.logger.dev.info("[dmNostrService] Scheduling relay reconnect", {
      relayUrl,
      reason,
      backoffMs,
    });

    const timer = setTimeout(() => {
      this.relayTimers.delete(relayUrl);
      this.resubscribeRelay(relayUrl);
    }, backoffMs);

    this.relayTimers.set(relayUrl, timer);
  }

  buildFilters({ actorPubkey, authors = [], recipients = [] } = {}) {
    const normalizedActor = normalizePubkey(actorPubkey || this.actorPubkey);
    const authorSet = new Set(
      [normalizedActor, ...authors].map(normalizePubkey).filter(Boolean),
    );
    const recipientSet = new Set(
      [normalizedActor, ...recipients].map(normalizePubkey).filter(Boolean),
    );

    const filters = [];

    if (authorSet.size) {
      filters.push({
        kinds: [4],
        authors: Array.from(authorSet),
      });
    }

    if (recipientSet.size) {
      filters.push({
        kinds: [4],
        "#p": Array.from(recipientSet),
      });
    }

    return filters;
  }

  updateWatermark(relayUrl, event) {
    if (!event || typeof event.created_at !== "number") {
      return;
    }

    const current = this.relayWatermarks.get(relayUrl) || 0;
    if (event.created_at > current) {
      this.relayWatermarks.set(relayUrl, event.created_at);
    }
  }

  hasSeenEvent(eventId) {
    if (!eventId) {
      return false;
    }
    return this.seenEventIds.has(eventId);
  }

  rememberEvent(eventId) {
    if (!eventId || this.seenEventIds.has(eventId)) {
      return;
    }

    this.seenEventIds.add(eventId);
    this.seenEventOrder.push(eventId);

    if (this.seenEventOrder.length > MAX_SEEN_EVENT_IDS) {
      const oldest = this.seenEventOrder.shift();
      if (oldest) {
        this.seenEventIds.delete(oldest);
      }
    }
  }

  shouldProcessEvent(relayUrl, event) {
    if (!event || typeof event.id !== "string") {
      return false;
    }

    if (this.hasSeenEvent(event.id)) {
      return false;
    }

    const watermark = this.relayWatermarks.get(relayUrl) || 0;
    if (typeof event.created_at === "number" && event.created_at < watermark) {
      return false;
    }

    return true;
  }

  async subscribe({
    actorPubkey,
    authors,
    recipients,
    relays,
    onEvent,
    onEose,
    onError,
  } = {}) {
    if (typeof onEvent === "function") {
      this.handlers.onEvent = onEvent;
    }
    if (typeof onEose === "function") {
      this.handlers.onEose = onEose;
    }
    if (typeof onError === "function") {
      this.handlers.onError = onError;
    }

    if (typeof actorPubkey === "string" && actorPubkey.trim()) {
      this.actorPubkey = normalizePubkey(actorPubkey);
    }

    if (Array.isArray(relays)) {
      this.relays = normalizeRelayList(relays);
    }

    if (!this.relays.length) {
      this.logger.user.warn("[dmNostrService] Cannot subscribe without relays.");
      return;
    }

    this.activeFilters = this.buildFilters({
      actorPubkey: this.actorPubkey,
      authors,
      recipients,
    });

    if (!this.activeFilters.length) {
      this.logger.user.warn("[dmNostrService] Cannot subscribe without filters.");
      return;
    }

    await this.connect();

    this.relays.forEach((relayUrl) => {
      this.subscribeRelay(relayUrl);
    });
  }

  subscribeRelay(relayUrl) {
    if (!this.pool || !relayUrl) {
      return;
    }

    const since = this.relayWatermarks.get(relayUrl);
    const filters = this.activeFilters.map((filter) => {
      if (typeof since === "number" && since > 0) {
        return {
          ...filter,
          since: Math.max(0, since - 1),
        };
      }
      return { ...filter };
    });

    if (this.subscriptions.has(relayUrl)) {
      this.unsubscribeRelay(relayUrl);
    }

    let subscription;
    try {
      subscription = this.pool.sub([relayUrl], filters);
    } catch (error) {
      this.recordRelayFailure(relayUrl, error);
      this.scheduleReconnect(relayUrl, "subscribe-failed");
      return;
    }

    if (!subscription || typeof subscription.on !== "function") {
      this.recordRelayFailure(relayUrl, new Error("subscription-unavailable"));
      this.scheduleReconnect(relayUrl, "subscribe-unavailable");
      return;
    }

    subscription.on("event", (event) => {
      if (!this.shouldProcessEvent(relayUrl, event)) {
        return;
      }

      this.rememberEvent(event.id);
      this.updateWatermark(relayUrl, event);
      this.recordRelaySuccess(relayUrl);

      if (typeof this.handlers.onEvent === "function") {
        try {
          this.handlers.onEvent(event, { relayUrl });
        } catch (error) {
          this.logger.dev.warn("[dmNostrService] onEvent handler failed.", error);
        }
      }
    });

    subscription.on("eose", () => {
      if (typeof this.handlers.onEose === "function") {
        try {
          this.handlers.onEose({ relayUrl });
        } catch (error) {
          this.logger.dev.warn("[dmNostrService] onEose handler failed.", error);
        }
      }
    });

    subscription.on("close", () => {
      this.recordRelayFailure(relayUrl, new Error("subscription-closed"));
      this.scheduleReconnect(relayUrl, "subscription-closed");
    });

    this.subscriptions.set(relayUrl, subscription);
  }

  resubscribeRelay(relayUrl) {
    if (!relayUrl) {
      return;
    }

    this.subscribeRelay(relayUrl);
  }

  unsubscribeRelay(relayUrl) {
    const subscription = this.subscriptions.get(relayUrl);
    if (subscription && typeof subscription.unsub === "function") {
      try {
        subscription.unsub();
      } catch (error) {
        this.logger.dev.warn("[dmNostrService] Failed to unsubscribe relay.", error);
      }
    }
    this.subscriptions.delete(relayUrl);
  }

  stop() {
    for (const relayUrl of this.subscriptions.keys()) {
      this.unsubscribeRelay(relayUrl);
    }

    for (const timer of this.relayTimers.values()) {
      clearTimeout(timer);
    }

    this.relayTimers.clear();
  }

  async publishSignedEvent(signedEvent, { relays = null } = {}) {
    if (!signedEvent || typeof signedEvent !== "object") {
      throw new Error("signed-event-required");
    }

    if (signedEvent.kind !== 4) {
      throw new Error("invalid-event-kind");
    }

    const relayList = normalizeRelayList(
      Array.isArray(relays) && relays.length
        ? relays
        : this.preferredRelays.length
        ? this.preferredRelays
        : this.relays,
    );

    if (!relayList.length) {
      throw new Error("no-relays-configured");
    }

    await this.ensurePool();

    const results = await Promise.all(
      relayList.map(async (relayUrl) => {
        let result;
        try {
          result = await publishEventToRelay(this.pool, relayUrl, signedEvent);
        } catch (error) {
          this.recordRelayFailure(relayUrl, error);
          return { success: false, error, relayUrl };
        }

        if (result?.success) {
          this.recordRelaySuccess(relayUrl);
        } else {
          this.recordRelayFailure(
            relayUrl,
            result?.error || new Error("publish-failed"),
          );
        }

        return { ...result, relayUrl };
      }),
    );

    return results;
  }
}

export default DmNostrService;