another.im-ios/ConversationsClassic/AppData/Store/ClientsStore.swift

import Combine
import Foundation
import GRDB

@MainActor
final class ClientsStore: ObservableObject {
    static let shared = ClientsStore()

    @Published private(set) var ready = false
    @Published private(set) var clients: [Client] = []
    @Published private(set) var actualRosters: [Roster] = []
    @Published private(set) var actualChats: [Chat] = []

    private var credentialsCancellable: AnyCancellable?
    private var rostersCancellable: AnyCancellable?
    private var chatsCancellable: AnyCancellable?

    init() {
        credentialsCancellable = ValueObservation
            .tracking { db in
                try Credentials.fetchAll(db)
            }
            .publisher(in: Database.shared.dbQueue)
            .catch { _ in Just([]) }
            .sink { [weak self] creds in
                self?.processCredentials(creds)
            }
    }

    private func processCredentials(_ credentials: [Credentials]) {
        let existsJids = Set(clients.map { $0.credentials.bareJid })
        let credentialsJids = Set(credentials.map { $0.bareJid })

        let forAdd = credentials.filter { !existsJids.contains($0.bareJid) }
        let newClients = forAdd.map { Client(credentials: $0) }

        let forRemove = clients.filter { !credentialsJids.contains($0.credentials.bareJid) }
        forRemove.forEach { $0.disconnect() }

        var updatedClients = clients.filter { credentialsJids.contains($0.credentials.bareJid) }
        updatedClients.append(contentsOf: newClients)
        clients = updatedClients

        if !ready {
            ready = true
        }

        resubscribeRosters()
        resubscribeChats()
        reconnectAll()
    }

    private func client(for credentials: Credentials) -> Client? {
        clients.first { $0.credentials == credentials }
    }
}

extension ClientsStore {
    func tryLogin(_ jidStr: String, _ pass: String) async throws {
        // login with fake timeout
        async let sleep: Void? = try? await Task.sleep(nanoseconds: 1 * NSEC_PER_SEC)
        async let request = try await Client.tryLogin(with: .init(bareJid: jidStr, pass: pass, isActive: true))
        let client = try await(request, sleep).0

        clients.append(client)
        try? await client.credentials.save()
    }

    private func reconnectAll() {
        Task {
            await withTaskGroup(of: Void.self) { taskGroup in
                for client in clients {
                    taskGroup.addTask {
                        await client.connect()
                    }
                }
            }
        }
    }
}

extension ClientsStore {
    private func resubscribeRosters() {
        let clientsJids = clients
            .filter { $0.state != .disabled }
            .map { $0.credentials.bareJid }

        rostersCancellable = ValueObservation.tracking { db in
            try Roster
                .filter(clientsJids.contains(Column("bareJid")))
                .filter(Column("locallyDeleted") == false)
                .fetchAll(db)
        }
        .publisher(in: Database.shared.dbQueue)
        .catch { _ in Just([]) }
        .sink { [weak self] rosters in
            self?.actualRosters = rosters
        }
    }

    func addRoster(_ credentials: Credentials, contactJID: String, name: String?, groups: [String]) async throws {
        // check that roster exist in db as locally deleted and undelete it
        let deletedLocally = try await Roster.fetchDeletedLocally()
        if var roster = deletedLocally.first(where: { $0.contactBareJid == contactJID }) {
            try await roster.setLocallyDeleted(false)
            return
        }

        // add new roster
        guard let client = client(for: credentials) else {
            throw ClientStoreError.clientNotFound
        }
        try await client.addRoster(contactJID, name: name, groups: groups)
    }

    func deleteRoster(_ roster: Roster) async throws {
        guard let client = clients.first(where: { $0.credentials.bareJid == roster.bareJid }) else {
            throw ClientStoreError.clientNotFound
        }
        try await client.deleteRoster(roster)
    }
}

extension ClientsStore {
    private func resubscribeChats() {
        let clientsJids = clients
            .filter { $0.state != .disabled }
            .map { $0.credentials.bareJid }

        chatsCancellable = ValueObservation.tracking { db in
            try Chat
                .filter(clientsJids.contains(Column("account")))
                .fetchAll(db)
        }
        .publisher(in: Database.shared.dbQueue)
        .catch { _ in Just([]) }
        .sink { [weak self] chats in
            self?.actualChats = chats
        }
    }
}

extension ClientsStore {
    func conversation(for roster: Roster) async throws -> ConversationStore {
        while !ready {
            await Task.yield()
        }

        guard let client = clients.first(where: { $0.credentials.bareJid == roster.bareJid }) else {
            throw ClientStoreError.clientNotFound
        }

        return ConversationStore(roster: roster, client: client)
    }

    func conversation(for chat: Chat) async throws -> ConversationStore {
        while !ready {
            await Task.yield()
        }

        guard let client = clients.first(where: { $0.credentials.bareJid == chat.account }) else {
            throw ClientStoreError.clientNotFound
        }

        let roster = try await chat.fetchRoster()
        return ConversationStore(roster: roster, client: client)
    }
}