plezy/lib/connection/connection_registry.dart

import 'dart:async';
import 'dart:convert';

import 'package:collection/collection.dart';
import 'package:drift/drift.dart';

import '../database/app_database.dart';
import '../media/media_backend.dart';
import '../services/credential_vault.dart';
import '../utils/app_logger.dart';
import 'connection.dart';

/// CRUD over the persisted [Connections] table. The registry is the source
/// of truth for which connections the user has added; the runtime
/// `MultiServerManager` populates per-server clients from these records.
class ConnectionRegistry {
  ConnectionRegistry(this._db);

  final AppDatabase _db;
  static const DeepCollectionEquality _configEquality = DeepCollectionEquality();

  final Expando<Map<String, Object?>> _decryptedConfigs = Expando<Map<String, Object?>>(
    'ConnectionRegistry.decryptedConfigs',
  );

  /// Compares the decrypted persisted config projections retained while rows
  /// are decoded. The fallback covers models supplied outside this registry.
  bool hasSameConfig(Connection a, Connection b) {
    final aConfig = _decryptedConfigs[a] ??= a.toConfigJson();
    final bConfig = _decryptedConfigs[b] ??= b.toConfigJson();
    return _configEquality.equals(aConfig, bConfig);
  }

  /// Emits the current set of connections after every mutation. Drift's
  /// `watch()` provides this for free.
  Stream<List<Connection>> watchConnections() {
    return (_db.select(_db.connections)..orderBy([(t) => OrderingTerm.asc(t.createdAt)])).watch().asyncMap(
      (rows) async => (await Future.wait(rows.map(_rowToConnection))).whereType<Connection>().toList(),
    );
  }

  /// One-shot fetch of all stored connections.
  Future<List<Connection>> list() async {
    final rows = await (_db.select(_db.connections)..orderBy([(t) => OrderingTerm.asc(t.createdAt)])).get();
    return (await Future.wait(rows.map(_rowToConnection))).whereType<Connection>().toList();
  }

  /// Lookup a connection by id.
  Future<Connection?> get(String id) async {
    final row = await (_db.select(_db.connections)..where((t) => t.id.equals(id))).getSingleOrNull();
    if (row == null) return null;
    return _rowToConnection(row);
  }

  /// Insert or replace [connection]. Re-upserting an existing row keeps the
  /// row's current `createdAt` (so token/metadata refreshes don't restamp
  /// creation order).
  ///
  /// Creation order is behaviour, not bookkeeping: it decides which
  /// connection lends a profile its picture. Re-authenticating rebuilds the
  /// model with `DateTime.now()` and reuses the same stable id, so without
  /// this the originally-first connection would jump to last on every
  /// re-sign-in.
  Future<void> upsert(Connection connection) async {
    await _db.runIdentityMutation(() async {
      final existing = await (_db.select(_db.connections)..where((t) => t.id.equals(connection.id))).getSingleOrNull();
      final createdAt = existing?.createdAt ?? connection.createdAt.millisecondsSinceEpoch;
      final protectedConfig = await CredentialVault.protectConnectionConfig(
        connection.kind.id,
        connection.toConfigJson(),
      );
      final row = ConnectionsCompanion(
        id: Value(connection.id),
        kind: Value(connection.kind.id),
        displayName: Value(connection.displayName),
        configJson: Value(jsonEncode(protectedConfig)),
        createdAt: Value(createdAt),
        lastAuthenticatedAt: Value(connection.lastAuthenticatedAt?.millisecondsSinceEpoch),
      );
      await _db.into(_db.connections).insertOnConflictUpdate(row);
    });
    appLogger.d('ConnectionRegistry: upserted ${connection.kind.id}/${connection.id}');
  }

  /// Remove a stored connection.
  Future<void> remove(String id) async {
    await _db.runIdentityMutation(() async {
      await (_db.delete(_db.connections)..where((t) => t.id.equals(id))).go();
    });
    appLogger.d('ConnectionRegistry: removed $id');
  }

  Future<void> clear() async {
    await _db.runIdentityMutation(() async {
      await _db.delete(_db.connections).go();
    });
  }

  /// All Plex accounts in insertion order. Convenience over
  /// `(await list()).whereType<PlexAccountConnection>()` — cuts ~3 lines from
  /// every caller that needs to filter by backend.
  Future<List<PlexAccountConnection>> listPlexAccounts() async {
    final all = await list();
    return all.whereType<PlexAccountConnection>().toList();
  }

  /// Lookup a [PlexAccountConnection] by id. Returns `null` if no row
  /// matches OR the row exists but isn't a Plex account.
  Future<PlexAccountConnection?> getPlexAccount(String id) async {
    final c = await get(id);
    return c is PlexAccountConnection ? c : null;
  }

  Future<Connection?> _rowToConnection(ConnectionRow row) async {
    try {
      final json = jsonDecode(row.configJson) as Map<String, dynamic>;
      final kind = MediaBackend.fromId(row.kind);
      final revealed = await CredentialVault.revealConnectionConfig(kind.id, json);
      final createdAt = DateTime.fromMillisecondsSinceEpoch(row.createdAt);
      final lastAuth = row.lastAuthenticatedAt == null
          ? null
          : DateTime.fromMillisecondsSinceEpoch(row.lastAuthenticatedAt!);
      final connection = switch (kind) {
        MediaBackend.plex => PlexAccountConnection.fromConfigJson(
          id: row.id,
          json: revealed.config,
          createdAt: createdAt,
          lastAuthenticatedAt: lastAuth,
        ),
        MediaBackend.jellyfin || MediaBackend.emby => JellyfinConnection.fromConfigJson(
          id: row.id,
          json: revealed.config,
          createdAt: createdAt,
          lastAuthenticatedAt: lastAuth,
          dialect: kind.dialect!,
        ),
      };
      _decryptedConfigs[connection] = revealed.config;
      if (revealed.migrated) {
        await upsert(connection);
      }
      return connection;
    } catch (e, st) {
      appLogger.e('ConnectionRegistry: failed to decode connection ${row.id}', error: e, stackTrace: st);
      return null;
    }
  }
}