textplus/Sources/PostgresStore.swift

import Foundation
import CryptoKit

enum PGSettings {
    static let autoIngestKey = "pgAutoIngest"
    static let uriKey = "pgURI"
    static let defaultURI = "postgresql://localhost:5433/scraperrr"

    static var autoIngest: Bool { UserDefaults.standard.bool(forKey: autoIngestKey) }
    static var uri: String {
        let v = (UserDefaults.standard.string(forKey: uriKey) ?? "")
            .trimmingCharacters(in: .whitespacesAndNewlines)
        return v.isEmpty ? defaultURI : v
    }
}

struct PGError: Error {
    let message: String
}

/// Talks to Postgres by piping SQL into `psql` — no client library to link.
/// All values are escaped as SQL literals; numeric ids are validated in Swift.
enum PostgresStore {

    static var psqlPath: String? {
        let candidates = [
            "/opt/homebrew/opt/postgresql@17/bin/psql",
            "/opt/homebrew/opt/postgresql@18/bin/psql",
            "/opt/homebrew/opt/postgresql@16/bin/psql",
            "/opt/homebrew/bin/psql",
            "/usr/local/bin/psql",
            "/usr/bin/psql",
        ]
        return candidates.first { FileManager.default.isExecutableFile(atPath: $0) }
    }

    struct PageIngest {
        let url: String?
        let sourceName: String
        let contentHash: String
        let elementCount: Int
        let selectors: [(selector: String, count: Int, sample: String)]
        let profileName: String?
        let header: [String]
        let rows: [[String]]
    }

    struct IngestReport {
        let pageId: Int64
        let selectorCount: Int
        let rowCount: Int
        let snapshotCount: Int
        let skippedRows: Int
        let skippedDuplicate: Bool
    }

    static func sha256(_ s: String) -> String {
        SHA256.hash(data: Data(s.utf8)).map { String(format: "%02x", $0) }.joined()
    }

    // MARK: - Public entry points (call from main; completions on main)

    static func test(uri: String, completion: @escaping (Result<String, PGError>) -> Void) {
        let sql = schemaSQL + "\nSELECT current_database() || ' · ' || split_part(version(), ' on ', 1);"
        runSQL(uri: uri, sql: sql) { result in
            switch result {
            case .success(let out):
                let last = out.split(separator: "\n").last.map(String.init) ?? "ok"
                completion(.success("Connected, tables ready — \(last)"))
            case .failure(let error):
                completion(.failure(error))
            }
        }
    }

    static func ingest(_ page: PageIngest, force: Bool,
                       completion: @escaping (Result<IngestReport, PGError>) -> Void) {
        let uri = PGSettings.uri

        func runMainIngest() {
            let built = buildIngestSQL(page)
            runSQL(uri: uri, sql: built.sql) { result in
                switch result {
                case .success(let out):
                    guard let idLine = out.split(separator: "\n").last,
                          let pageId = Int64(idLine.trimmingCharacters(in: .whitespaces)) else {
                        completion(.failure(PGError(message: "Couldn't read the new page id from psql output: \(out.suffix(120))")))
                        return
                    }
                    completion(.success(IngestReport(pageId: pageId,
                                                     selectorCount: page.selectors.count,
                                                     rowCount: page.rows.count,
                                                     snapshotCount: built.snapshotCount,
                                                     skippedRows: built.skipped,
                                                     skippedDuplicate: false)))
                case .failure(let error):
                    completion(.failure(error))
                }
            }
        }

        if force || page.url == nil {
            runMainIngest()
            return
        }
        // Auto mode: skip if the latest capture of this URL has identical content.
        let checkSQL = schemaSQL + """

        SELECT content_hash FROM tp_pages WHERE url = \(lit(page.url))
        ORDER BY captured_at DESC LIMIT 1;
        """
        runSQL(uri: uri, sql: checkSQL) { result in
            switch result {
            case .success(let out):
                let lastHash = out.split(separator: "\n").last.map(String.init) ?? ""
                if lastHash == page.contentHash {
                    completion(.success(IngestReport(pageId: -1, selectorCount: 0, rowCount: 0,
                                                     snapshotCount: 0, skippedRows: 0, skippedDuplicate: true)))
                } else {
                    runMainIngest()
                }
            case .failure(let error):
                completion(.failure(error))
            }
        }
    }

    // MARK: - SQL construction

    private static let schemaSQL = """
    CREATE TABLE IF NOT EXISTS tp_pages (
        id BIGSERIAL PRIMARY KEY,
        url TEXT,
        source_name TEXT,
        content_hash TEXT,
        element_count INT,
        selector_count INT,
        captured_at TIMESTAMPTZ DEFAULT now()
    );
    CREATE TABLE IF NOT EXISTS tp_selectors (
        id BIGSERIAL PRIMARY KEY,
        page_id BIGINT REFERENCES tp_pages(id) ON DELETE CASCADE,
        selector TEXT NOT NULL,
        match_count INT NOT NULL,
        sample TEXT
    );
    CREATE TABLE IF NOT EXISTS tp_rows (
        id BIGSERIAL PRIMARY KEY,
        page_id BIGINT REFERENCES tp_pages(id) ON DELETE CASCADE,
        profile TEXT,
        row_num INT,
        fields JSONB
    );
    CREATE INDEX IF NOT EXISTS idx_tp_pages_url ON tp_pages(url);
    CREATE INDEX IF NOT EXISTS idx_tp_rows_page ON tp_rows(page_id);
    CREATE TABLE IF NOT EXISTS discogs_sellers (
        seller_id BIGINT PRIMARY KEY,
        username VARCHAR(100) NOT NULL,
        last_scraped_at TIMESTAMPTZ DEFAULT now()
    );
    -- A username can be recycled across seller_ids over time; the seller_id is
    -- the real key. Drop the old UNIQUE(username) so a reused name can't abort
    -- a whole page ingest. (IF NOT EXISTS skips the table on existing installs,
    -- so the constraint must be dropped explicitly.)
    ALTER TABLE discogs_sellers DROP CONSTRAINT IF EXISTS discogs_sellers_username_key;
    CREATE TABLE IF NOT EXISTS marketplace_snapshots (
        item_id BIGINT PRIMARY KEY,
        release_id BIGINT NOT NULL,
        seller_id BIGINT REFERENCES discogs_sellers(seller_id),
        title TEXT,
        listing_url TEXT,
        media_condition VARCHAR(120),
        sleeve_condition VARCHAR(120),
        price_currency VARCHAR(10),
        price_value NUMERIC(12,2),
        shipping_value NUMERIC(12,2),
        first_seen TIMESTAMPTZ DEFAULT now(),
        last_seen TIMESTAMPTZ DEFAULT now(),
        is_active BOOLEAN DEFAULT TRUE
    );
    CREATE INDEX IF NOT EXISTS idx_market_release_id ON marketplace_snapshots(release_id);
    CREATE INDEX IF NOT EXISTS idx_market_seller_active ON marketplace_snapshots(seller_id) WHERE is_active = TRUE;
    """

    private static func buildIngestSQL(_ page: PageIngest) -> (sql: String, snapshotCount: Int, skipped: Int) {
        var sql = schemaSQL + "\nBEGIN;\n"
        sql += """
        INSERT INTO tp_pages (url, source_name, content_hash, element_count, selector_count)
        VALUES (\(lit(page.url)), \(lit(page.sourceName)), \(lit(page.contentHash)), \(page.elementCount), \(page.selectors.count))
        RETURNING id AS page_id \\gset

        """

        for chunk in page.selectors.chunked(into: 500) {
            let values = chunk.map {
                "(:page_id, \(lit($0.selector)), \($0.count), \(lit(String($0.sample.prefix(300)))))"
            }.joined(separator: ",\n")
            sql += "INSERT INTO tp_selectors (page_id, selector, match_count, sample) VALUES\n\(values);\n"
        }

        if !page.rows.isEmpty, let profile = page.profileName {
            for (chunkIndex, chunk) in page.rows.chunked(into: 200).enumerated() {
                let values = chunk.enumerated().map { offset, row -> String in
                    let rowNum = chunkIndex * 200 + offset + 1
                    var dict: [String: String] = [:]
                    for (i, name) in page.header.enumerated() where i < row.count {
                        // Condition fields capture the grade plus its tooltip
                        // description; keep the JSONB mirror as clean as the
                        // snapshot column (grade through the closing paren).
                        if name.hasSuffix("_condition"), let close = row[i].range(of: ")") {
                            dict[name] = String(row[i][..<close.upperBound])
                        } else {
                            dict[name] = row[i]
                        }
                    }
                    let json = (try? JSONSerialization.data(withJSONObject: dict))
                        .flatMap { String(data: $0, encoding: .utf8) } ?? "{}"
                    return "(:page_id, \(lit(profile)), \(rowNum), \(lit(json))::jsonb)"
                }.joined(separator: ",\n")
                sql += "INSERT INTO tp_rows (page_id, profile, row_num, fields) VALUES\n\(values);\n"
            }
        }

        var snapshotCount = 0
        var skipped = 0
        if page.profileName == ScrapeProfile.discogsMarketplace.name {
            let upserts = buildDiscogsUpserts(header: page.header, rows: page.rows)
            sql += upserts.sql
            snapshotCount = upserts.snapshotCount
            skipped = upserts.skipped
        }

        sql += "COMMIT;\nSELECT :page_id;\n"
        return (sql, snapshotCount, skipped)
    }

    /// Maps profile rows onto the scraperrr snapshot schema:
    /// sellers upserted by data-seller-id, listings upserted by data-item-id
    /// with last_seen/price refreshed on re-capture.
    private static func buildDiscogsUpserts(header: [String], rows: [[String]]) -> (sql: String, snapshotCount: Int, skipped: Int) {
        func col(_ name: String) -> Int? { header.firstIndex(of: name) }
        guard let itemIdx = col("item_id"), let relIdx = col("release_id") else { return ("", 0, 0) }
        let sellerIdIdx = col("seller_id"), sellerNameIdx = col("seller_username")
        let sellerNameAltIdx = col("seller_username_alt")
        let titleIdx = col("title"), urlIdx = col("listing_url")
        let mediaIdx = col("media_condition"), sleeveIdx = col("sleeve_condition")
        let curIdx = col("price_currency"), priceIdx = col("price_value")
        let shipIdx = col("shipping")

        func value(_ row: [String], _ idx: Int?) -> String? {
            guard let idx, idx < row.count else { return nil }
            let v = row[idx].trimmingCharacters(in: .whitespaces)
            return v.isEmpty ? nil : v
        }

        // "Near Mint (NM or M-) A nearly perfect record…" → keep through the ")".
        func condition(_ s: String?) -> String? {
            guard let s else { return nil }
            if let close = s.range(of: ")") { return String(s[..<close.upperBound]) }
            return String(s.prefix(60))
        }

        func numeric(_ s: String?) -> String? {
            guard let s, let d = Double(s) else { return nil }
            return String(format: "%.2f", d)
        }

        func firstDecimal(_ s: String?) -> String? {
            guard let s, let r = s.range(of: #"\d+(\.\d+)?"#, options: .regularExpression) else { return nil }
            return numeric(String(s[r]))
        }

        var sellers: [Int64: String] = [:]
        var snapshots: [Int64: String] = [:]  // item_id → VALUES tuple (dedup within batch)
        var skipped = 0

        for row in rows {
            guard let itemId = value(row, itemIdx).flatMap({ Int64($0) }),
                  let releaseId = value(row, relIdx).flatMap({ Int64($0) }) else {
                skipped += 1
                continue
            }
            // Only reference a seller_id the snapshot's FK can resolve: upsert
            // the seller whenever its id is present (falling back to the
            // co-located data-username, then a derived name), and NULL the
            // snapshot's seller_id when no id was found.
            var linkedSellerId: Int64?
            if let sellerId = value(row, sellerIdIdx).flatMap({ Int64($0) }) {
                let name = value(row, sellerNameIdx)
                    ?? value(row, sellerNameAltIdx)
                    ?? "seller_\(sellerId)"
                sellers[sellerId] = name
                linkedSellerId = sellerId
            }
            let tuple = """
            (\(itemId), \(releaseId), \(linkedSellerId.map(String.init) ?? "NULL"), \
            \(lit(value(row, titleIdx))), \(lit(value(row, urlIdx))), \
            \(lit(condition(value(row, mediaIdx)))), \(lit(condition(value(row, sleeveIdx)))), \
            \(lit(value(row, curIdx))), \(numeric(value(row, priceIdx)) ?? "NULL"), \
            \(firstDecimal(value(row, shipIdx)) ?? "NULL"))
            """
            snapshots[itemId] = tuple
        }
        guard !snapshots.isEmpty else { return ("", 0, skipped) }

        var sql = ""
        if !sellers.isEmpty {
            let sellerValues = sellers
                .sorted { $0.key < $1.key }
                .map { "(\($0.key), \(lit(String($0.value.prefix(100)))))" }
                .joined(separator: ",\n")
            sql += """
            INSERT INTO discogs_sellers (seller_id, username) VALUES
            \(sellerValues)
            ON CONFLICT (seller_id) DO UPDATE SET username = EXCLUDED.username, last_scraped_at = now();

            """
        }
        let snapshotValues = snapshots.sorted { $0.key < $1.key }.map { $0.value }.joined(separator: ",\n")
        sql += """
        INSERT INTO marketplace_snapshots
            (item_id, release_id, seller_id, title, listing_url, media_condition,
             sleeve_condition, price_currency, price_value, shipping_value) VALUES
        \(snapshotValues)
        ON CONFLICT (item_id) DO UPDATE SET
            last_seen = now(),
            price_value = EXCLUDED.price_value,
            shipping_value = EXCLUDED.shipping_value,
            media_condition = EXCLUDED.media_condition,
            sleeve_condition = EXCLUDED.sleeve_condition,
            is_active = TRUE;

        """
        return (sql, snapshots.count, skipped)
    }

    private static func lit(_ s: String?) -> String {
        guard let s else { return "NULL" }
        let cleaned = s.replacingOccurrences(of: "\0", with: "")
        return "'" + cleaned.replacingOccurrences(of: "'", with: "''") + "'"
    }

    // MARK: - psql runner

    private static let workQueue = DispatchQueue(label: "textplus.postgres")
    private static let timeoutSeconds: TimeInterval = 60

    private static func runSQL(uri: String, sql: String,
                               completion: @escaping (Result<String, PGError>) -> Void) {
        guard let psql = psqlPath else {
            DispatchQueue.main.async {
                completion(.failure(PGError(message: "psql not found — install with: brew install postgresql@17")))
            }
            return
        }
        workQueue.async {
            let process = Process()
            process.executableURL = URL(fileURLWithPath: psql)
            process.arguments = ["-X", "-q", "-A", "-t", "-v", "ON_ERROR_STOP=1", uri]

            let inPipe = Pipe(), outPipe = Pipe(), errPipe = Pipe()
            process.standardInput = inPipe
            process.standardOutput = outPipe
            process.standardError = errPipe

            do {
                try process.run()
            } catch {
                DispatchQueue.main.async { completion(.failure(PGError(message: error.localizedDescription))) }
                return
            }

            let watchdog = DispatchWorkItem { [weak process] in
                guard let process, process.isRunning else { return }
                process.terminate()
            }
            DispatchQueue.global().asyncAfter(deadline: .now() + timeoutSeconds, execute: watchdog)

            // Drain stdout/stderr concurrently while we feed stdin, so a large
            // script can't deadlock on a full pipe in either direction.
            var outData = Data()
            var errData = Data()
            let group = DispatchGroup()
            group.enter()
            DispatchQueue.global(qos: .utility).async {
                outData = outPipe.fileHandleForReading.readDataToEndOfFile()
                group.leave()
            }
            group.enter()
            DispatchQueue.global(qos: .utility).async {
                errData = errPipe.fileHandleForReading.readDataToEndOfFile()
                group.leave()
            }

            let writer = inPipe.fileHandleForWriting
            do {
                try writer.write(contentsOf: Data(sql.utf8))
                try writer.close()
            } catch {
                // Broken pipe (psql died early) — fall through to read its stderr.
                try? writer.close()
            }

            group.wait()
            process.waitUntilExit()
            watchdog.cancel()

            let out = String(decoding: outData, as: UTF8.self).trimmingCharacters(in: .whitespacesAndNewlines)
            let err = String(decoding: errData, as: UTF8.self).trimmingCharacters(in: .whitespacesAndNewlines)
            let status = process.terminationStatus

            DispatchQueue.main.async {
                if status == 0 {
                    completion(.success(out))
                } else {
                    let detail = err.isEmpty ? "psql exited with status \(status)" : err
                    completion(.failure(PGError(message: detail)))
                }
            }
        }
    }
}

private extension Array {
    func chunked(into size: Int) -> [[Element]] {
        stride(from: 0, to: count, by: size).map { Array(self[$0..<Swift.min($0 + size, count)]) }
    }
}