archivdms/internal/storage/accounting_pull.go

// Read-only query layer for the Buchhaltungs-Pull-API (see
// internal/api/accounting_handlers.go). Deliberately separate from
// documents.go's ListDocuments: the accounting view is a machine-to-machine
// export with its own reduced projection (no ocr_text, no storage_path, no
// content_hash) and keyset pagination over (created_at, id).
//
// Tenant isolation: every query here takes tenantID as its FIRST parameter and
// filters `WHERE d.tenant_id = $1` — there is no variant without it. The
// caller (the Bearer-auth middleware) derives that id solely from the resolved
// API key, never from the request.
//
// No permission-group ACL filter is applied: an accounting API key is a
// tenant-level machine credential (like the SFTP inbox account), not a user
// session. That is why creating one requires domain_admin.
package storage

import (
	"context"
	"encoding/base64"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/jackc/pgx/v5"
)

// ErrInvalidAccountingCursor is returned when a client-supplied cursor cannot
// be decoded. The handler maps this to HTTP 400.
var ErrInvalidAccountingCursor = errors.New("storage: invalid accounting cursor")

// AccountingDocument is the reduced, export-safe projection of a document for
// the pull API. storage_path / content_hash / ocr_text are intentionally
// absent (GoBD/security: the WORM location is never exposed; the file is only
// reachable through the streaming endpoint).
type AccountingDocument struct {
	ID                int64      `json:"id"`
	Title             string     `json:"title"`
	DocumentDate      *time.Time `json:"document_date,omitempty"`
	DocumentDateScore *float64   `json:"document_date_score,omitempty"`
	DocTypeID         *int64     `json:"doc_type_id,omitempty"`
	DocType           string     `json:"doc_type,omitempty"`
	CorrespondentID   *int64     `json:"correspondent_id,omitempty"`
	Correspondent     string     `json:"correspondent,omitempty"`
	CreatedAt         time.Time  `json:"created_at"`
}

// AccountingDocumentFilter holds the (already validated) query parameters of
// GET /api/v1/accounting/documents. TenantID is NOT part of it on purpose — it
// is passed separately from the API-key context so it can never be overwritten
// by a decoded request body/query.
type AccountingDocumentFilter struct {
	// Since/Until bound document_date (inclusive/exclusive respectively).
	Since *time.Time
	Until *time.Time
	// DocTypeID restricts to one document type.
	DocTypeID *int64
	// MinDateScore is the confidence quality gate (e.g. 0.75); documents with a
	// NULL score are excluded as soon as this is set.
	MinDateScore *float64
	// Cursor is the opaque keyset cursor from a previous page ("" = first page).
	Cursor string
	// Limit is the page size (already clamped by the handler).
	Limit int
}

// AccountingPage is one page of pull results plus the cursor for the next one.
type AccountingPage struct {
	Documents  []AccountingDocument `json:"documents"`
	NextCursor string               `json:"next_cursor,omitempty"`
	HasMore    bool                 `json:"has_more"`
}

// encodeAccountingCursor builds the opaque keyset cursor from the last row of a
// page. Format (base64url of) "<unix_nanos>:<id>" — the exact tuple the ORDER
// BY / WHERE comparison uses.
func encodeAccountingCursor(createdAt time.Time, id int64) string {
	raw := strconv.FormatInt(createdAt.UTC().UnixNano(), 10) + ":" + strconv.FormatInt(id, 10)
	return base64.RawURLEncoding.EncodeToString([]byte(raw))
}

// decodeAccountingCursor parses a cursor produced by encodeAccountingCursor.
func decodeAccountingCursor(cursor string) (time.Time, int64, error) {
	b, err := base64.RawURLEncoding.DecodeString(cursor)
	if err != nil {
		return time.Time{}, 0, ErrInvalidAccountingCursor
	}
	parts := strings.SplitN(string(b), ":", 2)
	if len(parts) != 2 {
		return time.Time{}, 0, ErrInvalidAccountingCursor
	}
	nanos, err := strconv.ParseInt(parts[0], 10, 64)
	if err != nil {
		return time.Time{}, 0, ErrInvalidAccountingCursor
	}
	id, err := strconv.ParseInt(parts[1], 10, 64)
	if err != nil {
		return time.Time{}, 0, ErrInvalidAccountingCursor
	}
	return time.Unix(0, nanos).UTC(), id, nil
}

// ListAccountingDocuments returns one keyset-paginated page of a tenant's
// documents for the pull API, ordered by (created_at, id) ascending so a
// consumer can poll incrementally without ever re-reading or skipping rows.
// Soft-deleted (trashed) documents are excluded.
func (s *Store) ListAccountingDocuments(ctx context.Context, tenantID int64, f AccountingDocumentFilter) (*AccountingPage, error) {
	limit := f.Limit
	if limit <= 0 {
		limit = 100
	}

	// $1 is always the tenant id — the isolation predicate is not optional.
	args := []any{tenantID}
	where := []string{"d.tenant_id = $1", "d.deleted_at IS NULL"}

	if f.Since != nil {
		args = append(args, *f.Since)
		where = append(where, fmt.Sprintf("d.document_date >= $%d", len(args)))
	}
	if f.Until != nil {
		args = append(args, *f.Until)
		where = append(where, fmt.Sprintf("d.document_date < $%d", len(args)))
	}
	if f.DocTypeID != nil {
		args = append(args, *f.DocTypeID)
		where = append(where, fmt.Sprintf("d.doc_type_id = $%d", len(args)))
	}
	if f.MinDateScore != nil {
		args = append(args, *f.MinDateScore)
		where = append(where, fmt.Sprintf("d.document_date_score IS NOT NULL AND d.document_date_score >= $%d", len(args)))
	}
	if f.Cursor != "" {
		curTS, curID, err := decodeAccountingCursor(f.Cursor)
		if err != nil {
			return nil, err
		}
		args = append(args, curTS, curID)
		where = append(where, fmt.Sprintf("(d.created_at, d.id) > ($%d, $%d)", len(args)-1, len(args)))
	}

	// Fetch one extra row to detect whether a further page exists.
	args = append(args, limit+1)
	query := `
		SELECT d.id, d.title, d.document_date, d.document_date_score,
			d.doc_type_id, COALESCE(dt.name, COALESCE(d.doc_type, '')),
			d.correspondent_id, COALESCE(c.name, COALESCE(d.correspondent, '')),
			d.created_at
		FROM documents d
		LEFT JOIN document_types dt ON dt.id = d.doc_type_id AND dt.tenant_id = d.tenant_id
		LEFT JOIN correspondents c ON c.id = d.correspondent_id AND c.tenant_id = d.tenant_id
		WHERE ` + strings.Join(where, " AND ") + `
		ORDER BY d.created_at ASC, d.id ASC
		LIMIT $` + strconv.Itoa(len(args))

	rows, err := s.db.Query(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("storage: list accounting documents: %w", err)
	}
	defer rows.Close()

	out := make([]AccountingDocument, 0, limit)
	for rows.Next() {
		var d AccountingDocument
		if err := rows.Scan(&d.ID, &d.Title, &d.DocumentDate, &d.DocumentDateScore,
			&d.DocTypeID, &d.DocType, &d.CorrespondentID, &d.Correspondent, &d.CreatedAt); err != nil {
			return nil, fmt.Errorf("storage: scan accounting document: %w", err)
		}
		out = append(out, d)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("storage: list accounting documents: %w", err)
	}

	page := &AccountingPage{Documents: out}
	if len(out) > limit {
		page.Documents = out[:limit]
		page.HasMore = true
		last := page.Documents[limit-1]
		page.NextCursor = encodeAccountingCursor(last.CreatedAt, last.ID)
	}
	return page, nil
}

// AccountingFileRef carries the server-side-only information needed to stream a
// document's WORM file. The storage path is unexported and reachable only via
// StoragePath(), mirroring storage.ResolvedShare, so a handler cannot
// accidentally serialise it into a response.
type AccountingFileRef struct {
	DocumentID  int64
	Title       string
	storagePath string
}

// StoragePath returns the WORM path of the file (server-side only).
func (r *AccountingFileRef) StoragePath() string { return r.storagePath }

// GetAccountingDocumentFile resolves a document's WORM file location, scoped to
// the tenant of the API key (id AND tenant_id — IDOR guard). Returns
// ErrDocumentNotFound for unknown id, foreign tenant and trashed document
// alike, so the endpoint never reveals whether a document exists outside the
// caller's tenant.
func (s *Store) GetAccountingDocumentFile(ctx context.Context, id, tenantID int64) (*AccountingFileRef, error) {
	var ref AccountingFileRef
	err := s.db.QueryRow(ctx, `
		SELECT id, title, storage_path
		FROM documents
		WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL
	`, id, tenantID).Scan(&ref.DocumentID, &ref.Title, &ref.storagePath)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, ErrDocumentNotFound
		}
		return nil, fmt.Errorf("storage: get accounting document file: %w", err)
	}
	return &ref, nil
}