query-orchestration/internal/database/migrations/00000000000102_document_views.up.sql

CREATE OR REPLACE FUNCTION listDocumentIDs(
    _clientId uuid,
    _batchSize INTEGER,
    _offset INTEGER
)
RETURNS TABLE (id uuid, totalCount BIGINT) AS $$
DECLARE
    totalCount BIGINT := 0;
BEGIN
    SELECT COUNT(*)::BIGINT INTO totalCount FROM documents WHERE clientId = _clientId;

    RETURN QUERY
        SELECT d.id, totalCount
        FROM documents as d
        WHERE d.clientId = _clientId
        ORDER BY d.id
        LIMIT _batchSize
        OFFSET _offset;
END;
$$ LANGUAGE plpgsql;

CREATE VIEW currentCleanEntries as
WITH RankedExtractions AS (
    SELECT
        dc.id,
        dc.documentId,
        dc.bucket,
        dc.key,
        dc.mimetype,
        dc.fail,
        dce.version,
        ROW_NUMBER() OVER (PARTITION BY dc.documentId ORDER BY dc.id DESC, dce.id DESC) as row_num
    FROM documents d
    JOIN currentCollectorMinCleanVersions ccv ON d.clientId = ccv.clientId
    JOIN documentCleans dc ON dc.documentId = d.id
    JOIN documentCleanEntries dce ON dc.id = dce.cleanId
        AND dce.version >= ccv.minCleanVersion
)
SELECT
    id,
    documentId,
    bucket,
    key,
    version,
    mimetype,
    fail
FROM RankedExtractions
WHERE row_num = 1;

CREATE VIEW currentTextEntries as
WITH RankedExtractions AS (
    SELECT
        dte.id,
        cc.documentId,
        dte.bucket,
        dte.key,
        dte.hash,
        dtee.version,
        dte.cleanEntryId,
        ROW_NUMBER() OVER (PARTITION BY cc.documentId ORDER BY dte.id DESC) as row_num
    FROM documents d
    JOIN currentCleanEntries cc on cc.documentId = d.id
    JOIN currentCollectorMinTextVersions ctv ON ctv.clientId = d.clientId
    JOIN documentTextExtractions dte ON dte.cleanEntryId = cc.id
    JOIN documentTextExtractionEntries dtee on dte.id = dtee.textId
        AND dtee.version >= ctv.minTextVersion
)
SELECT
    id,
    documentId,
    bucket,
    key,
    version,
    hash,
    cleanEntryId
FROM RankedExtractions
WHERE row_num = 1;

CREATE VIEW currentClientCanSync as
WITH RankedExtractions AS (
    SELECT
        ccs.clientId,
        ccs.canSync,
        ROW_NUMBER() OVER (PARTITION BY ccs.clientId ORDER BY ccs.id DESC) as row_num
    FROM clients c
    JOIN clientCanSync ccs on c.id = ccs.clientId
)
SELECT
    c.id as clientId,
    coalesce(r.canSync, false) as canSync
FROM clients c
LEFT JOIN RankedExtractions r on r.clientId = c.id and r.row_num = 1;

CREATE VIEW fullClients as
SELECT c.id, c.externalId, c.name, cs.canSync
    FROM clients as c
    JOIN currentClientCanSync as cs on cs.clientId = c.id;

CREATE OR REPLACE FUNCTION listValidDocumentResults(doc_id uuid)
RETURNS TABLE (documentId uuid, queryId uuid, resultId uuid, value text) AS $$
DECLARE
 count_before INT;
 count_after INT;
 iterations INT := 0;
BEGIN
    CREATE TEMPORARY TABLE allResults AS
    WITH
        docs AS (
            SELECT id as documentId, hash, clientId
            FROM documents
            WHERE id = doc_id
    ),
    query_dependency_tree AS (
        WITH RECURSIVE query_deps AS (
            SELECT
            q.queryId,
            unnest(q.requiredIds) AS requiredId
            FROM collectorQueryDependencyTree AS q
            JOIN docs AS d ON d.clientId = q.clientId
            WHERE array_length(q.requiredIds, 1) > 0

            UNION ALL

            SELECT
            qd.queryId,
            unnest(q.requiredIds) AS requiredId
            FROM query_deps qd
            JOIN collectorQueryDependencyTree q ON qd.requiredId = q.queryId
            WHERE array_length(q.requiredIds, 1) > 0
        )
        SELECT
        query_deps.queryId,
        array_agg(DISTINCT requiredId) AS all_required_ids
        FROM query_deps
        GROUP BY query_deps.queryId
    ),
    docResults AS (
    SELECT
        d.documentId,
        q.queryId,
        r.id AS resultId,
        r.value,
        q.requiredIds as directRequiredIds,
        COALESCE(qdt.all_required_ids, ARRAY[]::uuid[]) AS allRequiredIds
        FROM docs AS d
        JOIN collectorQueryDependencyTree AS q ON q.clientId = d.clientId
        LEFT JOIN query_dependency_tree qdt ON q.queryId = qdt.queryId
        LEFT JOIN currentTextEntries AS tte ON tte.documentId = d.documentId
        LEFT JOIN results AS r
            ON q.queryId = r.queryId
            AND r.queryVersion = q.queryVersion
            AND tte.id = r.textEntryId
        WHERE r.value is not null
    )
    SELECT * FROM docResults dr;

    CREATE TEMPORARY TABLE finalResults AS
    SELECT * from allResults where array_length(directRequiredIds, 1) is null;

    SELECT COUNT(*) INTO count_before FROM finalResults;
    RAISE NOTICE 'Starting with % rows', count_before;

    LOOP
        iterations := iterations + 1;

        INSERT INTO finalResults
            SELECT dr.*
            FROM allResults dr
            WHERE
                dr.directRequiredIds IS NOT NULL
                AND NOT EXISTS (
                    SELECT 1
                    FROM unnest(dr.directRequiredIds) AS req_id(queryId)
                    WHERE NOT EXISTS (
                        SELECT 1
                        FROM allResults req_dr
                        WHERE req_dr.queryId = req_id.queryId
                        AND req_dr.resultId IN (SELECT fr.resultId FROM finalResults fr)
                        AND exists (
                            select 1 from resultDependencies rrd
                            where dr.resultId = rrd.resultId
                            and req_dr.resultId = rrd.requiredResultId
                        )
                    )
                )
                and dr.resultId NOT IN (SELECT fr.resultId FROM finalResults fr);

        GET DIAGNOSTICS count_after = ROW_COUNT;
        RAISE NOTICE 'Iteration %: Additional Entries %',
        iterations, count_after;

        IF count_after = 0 THEN
            EXIT;
        END IF;
    END LOOP;

    RETURN QUERY SELECT tr.documentId, tr.queryId, tr.resultId, tr.value
        FROM finalResults tr;

    DROP TABLE finalResults;
    DROP TABLE allResults;
END;
$$ LANGUAGE plpgsql;