patrick
9a24ea29e1
Git-Repository für bestehenden archivdms-Code initialisiert, Branch-/Commit-Konvention (feature/<ticket>-<slug>-Branches, Ticket-Prefix in Commit-Nachricht) etabliert.
218 lines
6.8 KiB
Go
218 lines
6.8 KiB
Go
package storage
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
|
|
"archivdms/internal/index"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
)
|
|
|
|
// Full-text search index sync (Phase 1, see internal/index).
|
|
//
|
|
// Postgres is the single source of truth; these helpers keep the secondary
|
|
// per-tenant Manticore index in step. They are strictly best-effort: an index
|
|
// failure is logged and swallowed, NEVER returned to the caller, so a search
|
|
// backend hiccup can never block or fail a document write. When no indexer is
|
|
// configured (s.indexer == nil) every helper is a no-op.
|
|
|
|
// SyncIndex re-projects a document (with its tags + resolved ACL groups) into
|
|
// the search index. Safe to call after any change that affects an indexed
|
|
// field: create, tag attach/detach, doc_type/correspondent change, custom
|
|
// fields, or an ACL recompute. Best-effort — errors are logged, not returned.
|
|
func (s *Store) SyncIndex(ctx context.Context, documentID int64) {
|
|
if s.indexer == nil {
|
|
return
|
|
}
|
|
doc, err := s.buildDocumentDoc(ctx, documentID)
|
|
if err != nil {
|
|
s.logIndexWarn("build index doc", documentID, err)
|
|
return
|
|
}
|
|
if doc == nil {
|
|
// Row vanished (or tombstoned) — treat as a delete.
|
|
return
|
|
}
|
|
if err := s.indexer.ForTenant(doc.TenantID).IndexSync(ctx, *doc); err != nil {
|
|
s.logIndexWarn("index sync", documentID, err)
|
|
}
|
|
}
|
|
|
|
// DeleteFromIndex removes a document from the search index. Used on final
|
|
// (executed) deletion — GoBD-critical: a purged document must not remain
|
|
// findable. Best-effort — errors are logged, not returned. tenantID is passed
|
|
// explicitly because the DB row may already be gone/tombstoned.
|
|
func (s *Store) DeleteFromIndex(ctx context.Context, documentID, tenantID int64) {
|
|
if s.indexer == nil {
|
|
return
|
|
}
|
|
if err := s.indexer.ForTenant(tenantID).Delete(ctx, documentID); err != nil {
|
|
s.logIndexWarn("index delete", documentID, err)
|
|
}
|
|
}
|
|
|
|
// buildDocumentDoc assembles the index projection for a single document from
|
|
// the documents row plus its tags (document_tags/tags) and resolved ACL groups
|
|
// (document_visibility). Returns (nil, nil) if the document does not exist.
|
|
func (s *Store) buildDocumentDoc(ctx context.Context, documentID int64) (*index.DocumentDoc, error) {
|
|
var d index.DocumentDoc
|
|
err := s.db.QueryRow(ctx, `
|
|
SELECT id, tenant_id, title, COALESCE(doc_type, ''), COALESCE(correspondent, ''),
|
|
doc_type_id, correspondent_id, COALESCE(ocr_text, ''), retain_until, created_at, updated_at
|
|
FROM documents WHERE id = $1
|
|
`, documentID).Scan(&d.ID, &d.TenantID, &d.Title, &d.DocType, &d.Correspondent,
|
|
&d.DocTypeID, &d.CorrespondentID, &d.OCRText, &d.RetainUntil, &d.CreatedAt, &d.UpdatedAt)
|
|
if err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, nil
|
|
}
|
|
return nil, err
|
|
}
|
|
|
|
// Tags (names + ids), tenant-scoped via the join to tags.
|
|
tagRows, err := s.db.Query(ctx, `
|
|
SELECT t.id, t.name FROM tags t
|
|
JOIN document_tags dt ON dt.tag_id = t.id
|
|
WHERE dt.document_id = $1 AND t.tenant_id = $2
|
|
ORDER BY t.name ASC
|
|
`, documentID, d.TenantID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for tagRows.Next() {
|
|
var id int64
|
|
var name string
|
|
if err := tagRows.Scan(&id, &name); err != nil {
|
|
tagRows.Close()
|
|
return nil, err
|
|
}
|
|
d.TagIDs = append(d.TagIDs, id)
|
|
d.Tags = append(d.Tags, name)
|
|
}
|
|
tagRows.Close()
|
|
if err := tagRows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Resolved ACL groups (materialised visibility).
|
|
aclRows, err := s.db.Query(ctx, `
|
|
SELECT group_id FROM document_visibility WHERE document_id = $1 ORDER BY group_id
|
|
`, documentID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for aclRows.Next() {
|
|
var gid int64
|
|
if err := aclRows.Scan(&gid); err != nil {
|
|
aclRows.Close()
|
|
return nil, err
|
|
}
|
|
d.ACLGroupIDs = append(d.ACLGroupIDs, gid)
|
|
}
|
|
aclRows.Close()
|
|
if err := aclRows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return &d, nil
|
|
}
|
|
|
|
// ErrNoIndexer is returned by reindex helpers when no search index is wired
|
|
// into the store. Unlike the request-path sync helpers (which degrade to a
|
|
// silent no-op when s.indexer == nil), an explicit reindex must fail loudly so
|
|
// an operator never mistakes a no-op for a successful rebuild.
|
|
var ErrNoIndexer = errors.New("storage: no search index configured")
|
|
|
|
// ReindexTenant rebuilds the full-text search index for a single tenant from
|
|
// Postgres (the source of truth). It streams all non-deleted documents of the
|
|
// tenant in ascending-id batches (keyset pagination, batchSize rows at a time)
|
|
// so memory stays bounded even for very large tenants, projects each via
|
|
// buildDocumentDoc and upserts it through the tenant's Indexer.
|
|
//
|
|
// progress, if non-nil, is invoked after each successfully indexed document
|
|
// with (done, total) so callers can log progress. Returns the number of
|
|
// documents indexed. Fails with ErrNoIndexer when no indexer is configured.
|
|
func (s *Store) ReindexTenant(ctx context.Context, tenantID int64, batchSize int, progress func(done, total int)) (int, error) {
|
|
if s.indexer == nil {
|
|
return 0, ErrNoIndexer
|
|
}
|
|
if batchSize <= 0 {
|
|
batchSize = 500
|
|
}
|
|
|
|
var total int
|
|
if err := s.db.QueryRow(ctx,
|
|
`SELECT COUNT(*) FROM documents WHERE tenant_id = $1 AND deleted_at IS NULL`,
|
|
tenantID,
|
|
).Scan(&total); err != nil {
|
|
return 0, fmt.Errorf("storage: reindex count tenant %d: %w", tenantID, err)
|
|
}
|
|
|
|
indexer := s.indexer.ForTenant(tenantID)
|
|
|
|
done := 0
|
|
var lastID int64
|
|
for {
|
|
ids, err := s.reindexDocumentIDs(ctx, tenantID, lastID, batchSize)
|
|
if err != nil {
|
|
return done, err
|
|
}
|
|
if len(ids) == 0 {
|
|
break
|
|
}
|
|
for _, id := range ids {
|
|
doc, err := s.buildDocumentDoc(ctx, id)
|
|
if err != nil {
|
|
return done, fmt.Errorf("storage: reindex build doc id=%d: %w", id, err)
|
|
}
|
|
if doc == nil {
|
|
// Row vanished/tombstoned between the id scan and now — skip.
|
|
continue
|
|
}
|
|
if err := indexer.IndexSync(ctx, *doc); err != nil {
|
|
return done, fmt.Errorf("storage: reindex index doc id=%d: %w", id, err)
|
|
}
|
|
done++
|
|
if progress != nil {
|
|
progress(done, total)
|
|
}
|
|
}
|
|
lastID = ids[len(ids)-1]
|
|
}
|
|
|
|
return done, nil
|
|
}
|
|
|
|
// reindexDocumentIDs returns up to limit non-deleted document IDs for a tenant
|
|
// with id > afterID, ascending (keyset pagination).
|
|
func (s *Store) reindexDocumentIDs(ctx context.Context, tenantID, afterID int64, limit int) ([]int64, error) {
|
|
rows, err := s.db.Query(ctx, `
|
|
SELECT id FROM documents
|
|
WHERE tenant_id = $1 AND deleted_at IS NULL AND id > $2
|
|
ORDER BY id ASC
|
|
LIMIT $3
|
|
`, tenantID, afterID, limit)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("storage: reindex list ids tenant %d: %w", tenantID, err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
var ids []int64
|
|
for rows.Next() {
|
|
var id int64
|
|
if err := rows.Scan(&id); err != nil {
|
|
return nil, fmt.Errorf("storage: reindex scan id: %w", err)
|
|
}
|
|
ids = append(ids, id)
|
|
}
|
|
return ids, rows.Err()
|
|
}
|
|
|
|
func (s *Store) logIndexWarn(op string, documentID int64, err error) {
|
|
if s.logger != nil {
|
|
s.logger.Warn("index sync failed", "op", op, "document_id", documentID, "err", err)
|
|
}
|
|
}
|