From a9b4fa5eca28cbc89ec4cd8a5e1743a2c4e0129b Mon Sep 17 00:00:00 2001 From: napnap75 Date: Sun, 1 Nov 2026 23:02:24 +0100 Subject: [PATCH] archivist: initial Go implementation of the written user stories (US-01..US-23) --- archivist/go.mod | 12 + archivist/internal/config/config.go | 445 ++++++++++++++++++++++++ archivist/internal/config/render.go | 95 +++++ archivist/internal/pbapp/hooks.go | 200 +++++++++++ archivist/internal/pbapp/setup.go | 146 ++++++++ archivist/internal/pipeline/pipeline.go | 218 ++++++++++++ archivist/internal/storage/s3.go | 123 +++++++ archivist/main.go | 104 ++++++ 8 files changed, 1343 insertions(+) create mode 100644 archivist/go.mod create mode 100644 archivist/internal/config/config.go create mode 100644 archivist/internal/config/render.go create mode 100644 archivist/internal/pbapp/hooks.go create mode 100644 archivist/internal/pbapp/setup.go create mode 100644 archivist/internal/pipeline/pipeline.go create mode 100644 archivist/internal/storage/s3.go create mode 100644 archivist/main.go diff --git a/archivist/go.mod b/archivist/go.mod new file mode 100644 index 0000000..943bcba --- /dev/null +++ b/archivist/go.mod @@ -0,0 +1,12 @@ +module github.com/napnap75/archivist + +go 1.23 + +// Run `go mod tidy` to resolve the exact compatible versions. +require ( + github.com/expr-lang/expr v1.17.6 + github.com/ledongthuc/pdf v0.0.0-20240201144150-fd9d85c880aa + github.com/minio/minio-go/v7 v7.0.88 + github.com/pocketbase/pocketbase v0.26.6 + github.com/tailscale/hujson v0.4.0 +) diff --git a/archivist/internal/config/config.go b/archivist/internal/config/config.go new file mode 100644 index 0000000..e124923 --- /dev/null +++ b/archivist/internal/config/config.go @@ -0,0 +1,445 @@ +// Package config loads, validates and compiles the functional configuration +// (config.jsonc) per US-02. +package config + +import ( + "encoding/json" + "fmt" + "os" + "regexp" + "strconv" + "strings" + "time" + + "github.com/expr-lang/expr" + "github.com/tailscale/hujson" +) + +// Field is one declared extraction field. +type Field struct { + Name string `json:"name"` + Type string `json:"type"` // text | number | date | boolean | enum + Mandatory bool `json:"mandatory"` +} + +// Assignment is one "set" entry of a rule. +type Assignment struct { + Field string `json:"field"` + Value any `json:"value"` // constant (mutually exclusive with Eval) + Eval string `json:"eval"` // expression evaluated in the document context +} + +// Rule is one node of the ordered rule tree. +type Rule struct { + When string `json:"when"` // required expr expression + Set []Assignment `json:"set"` + Delete []string `json:"delete"` + Rules []*Rule `json:"rules"` +} + +type rawConfig struct { + Fields []Field `json:"fields"` + Rules []*Rule `json:"rules"` + Templates map[string]string `json:"templates"` +} + +// Config is the compiled functional configuration. +type Config struct { + Fields map[string]Field + FieldList []Field + Rules []*Rule + Templates map[string]*CompiledTemplate +} + +var validTypes = map[string]bool{"text": true, "number": true, "date": true, "boolean": true, "enum": true} + +// Load parses config.jsonc (comments allowed), validates it and compiles every +// regex and expression. Any error rejects the whole file (US-02, US-05). +func Load(path string) (*Config, error) { + data, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("read %s: %w", path, err) + } + // JSONC: standardize to plain JSON before unmarshalling. + std, err := hujson.Standardize(data) + if err != nil { + return nil, fmt.Errorf("parse %s: %w", path, err) + } + var raw rawConfig + if err := json.Unmarshal(std, &raw); err != nil { + return nil, fmt.Errorf("decode %s: %w", path, err) + } + + c := &Config{Fields: map[string]Field{}, Templates: map[string]*CompiledTemplate{}} + + // --- fields ----------------------------------------------------------- + seen := map[string]bool{} + for _, f := range raw.Fields { + if f.Name == "" { + return nil, fmt.Errorf("field with empty name") + } + if seen[f.Name] { + return nil, fmt.Errorf("duplicate field %q", f.Name) + } + if !validTypes[f.Type] { + return nil, fmt.Errorf("field %q: invalid type %q (want text|number|date|boolean|enum)", f.Name, f.Type) + } + seen[f.Name] = true + c.Fields[f.Name] = f + c.FieldList = append(c.FieldList, f) + } + + // --- rules -------------------------------------------------------------- + if err := validateRules(raw.Rules, c.Fields); err != nil { + return nil, err + } + c.Rules = raw.Rules + + // --- templates ----------------------------------------------------------- + for name, src := range raw.Templates { + ct, err := CompileTemplate(name, src) + if err != nil { + return nil, fmt.Errorf("template %q: %w", name, err) + } + c.Templates[name] = ct + } + + // Cross-reference check (US-02): template names used in the rules exist. + if err := checkRuleTemplates(raw.Rules, c); err != nil { + return nil, err + } + return c, nil +} + +func checkRuleTemplates(rules []*Rule, c *Config) error { + for _, r := range rules { + for _, s := range r.Set { + if s.Field == "template" { + name, _ := s.Value.(string) + if name != "" && s.Eval == "" { + if _, ok := c.Templates[name]; !ok { + return fmt.Errorf("rule sets template %q which does not exist in templates", name) + } + } + } + } + for _, child := range r.Rules { + if err := checkRuleTemplates(child, c); err != nil { + return err + } + } + } + return nil +} + +func validateRules(rules []*Rule, fields map[string]Field) error { + for _, r := range rules { + if r.When == "" { + return fmt.Errorf("rule with empty/missing when expression") + } + if _, err := CompileExpr(r.When); err != nil { + return fmt.Errorf("when expression: %w", err) + } + for _, s := range r.Set { + f, ok := fields[s.Field] + if !ok { + return fmt.Errorf("set: unknown field %q", s.Field) + } + if s.Eval != "" { + if _, err := CompileExpr(s.Eval); err != nil { + return fmt.Errorf("set %s: %w", s.Field, err) + } + } else if err := ValidateValue(f, s.Value); err != nil { + return fmt.Errorf("set %s: %w", s.Field, err) + } + } + for _, d := range r.Delete { + if _, ok := fields[d]; !ok { + return fmt.Errorf("delete: unknown field %q", d) + } + } + if err := validateRules(r.Rules, fields); err != nil { + return err + } + } + return nil +} + +// CompileExpr compiles one expression against the typed document context. +func CompileExpr(src string) (expr.Node, error) { + return expr.Compile(src, expr.Env(ExprEnv(nil, nil))) +} + +// ExprEnv builds the evaluation environment: pages + declared fields. +// Unset fields are exposed as nil so that references to them stay falsy +// instead of failing at runtime. +func ExprEnv(pages []string, fields map[string]any) map[string]any { + env := map[string]any{} + for _, f := range knownFields { + env[f] = nil + } + env["pages"] = pages + for k, v := range fields { + env[k] = v + } + return env +} + +var knownFields []string + +func SetKnownFields(fields []Field) { + knownFields = nil + for _, f := range fields { + knownFields = append(knownFields, f.Name) + } +} + +// ValidateValue checks that a rule-assigned value matches the declared type +// (type mismatch is a rule error, US-14). +func ValidateValue(f Field, v any) error { + switch f.Type { + case "text", "enum": + if _, ok := v.(string); !ok { + return fmt.Errorf("field %s (%s): value is not a string", f.Name, f.Type) + } + case "number": + switch v.(type) { + case float64, int, int64: + default: + return fmt.Errorf("field %s (number): value is not numeric", f.Name) + } + case "date": + if _, ok := v.(time.Time); !ok { + return fmt.Errorf("field %s (date): value is not a native date", f.Name) + } + case "boolean": + if _, ok := v.(bool); !ok { + return fmt.Errorf("field %s (boolean): value is not a boolean", f.Name) + } + } + return nil +} + +// Classify evaluates the ordered rule tree against the document pages and +// returns the extracted fields plus the list of rule errors (US-14). +func (c *Config) Classify(pages []string) (map[string]any, []string) { + fields := map[string]any{} + var errs []string + + var walk func(rules []*Rule) + walk = func(rules []*Rule) { + for _, r := range rules { + env := ExprEnv(pages, fields) + res, err := expr.Eval(r.When, env) + if err != nil { + errs = append(errs, fmt.Sprintf("when: %v", err)) + continue + } + if !truthy(res) { + continue + } + for _, s := range r.Set { + var v any + if s.Eval != "" { + var err error + v, err = expr.Eval(s.Eval, env) + if err != nil { + errs = append(errs, fmt.Sprintf("eval %s: %v", s.Field, err)) + continue + } + } else { + v = s.Value + } + if err := ValidateValue(c.Fields[s.Field], v); err != nil { + errs = append(errs, err.Error()) + continue + } + fields[s.Field] = v + } + for _, d := range r.Delete { + delete(fields, d) + } + walk(r.Rules) + } + } + walk(c.Rules) + return fields, errs +} + +func truthy(v any) bool { + switch t := v.(type) { + case nil: + return false + case bool: + return t + case string: + return t != "" + default: + return true + } +} + +// Evaluate runs classification and computes the verdict (US-15). +// It returns the fields, the rendered file name (empty when invalid) and the +// per-cause error list. +func (c *Config) Evaluate(pages []string) (map[string]any, string, []string) { + fields, errs := c.Classify(pages) + + // Mandatory fields. + for _, f := range c.FieldList { + if f.Mandatory { + if v, ok := fields[f.Name]; !ok || v == nil { + errs = append(errs, "mandatory field missing: "+f.Name) + } + } + } + + // Template selection: the "template" field, or "default". + tplName := "default" + if v, ok := fields["template"].(string); ok && v != "" { + tplName = v + } + tpl, ok := c.Templates[tplName] + if !ok { + errs = append(errs, "unknown template: "+tplName) + return fields, "", errs + } + rendered, err := tpl.Render(fields, pages) + if err != nil { + errs = append(errs, err.Error()) + return fields, "", errs + } + return fields, rendered, errs +} + +// FirstMatch is the firstMatch(regex, texts?) helper (US-14 / arch.md §5). +// It returns the first match (first capture group when present) or an empty +// string when no match is found. +func FirstMatch(args ...any) (any, error) { + if len(args) == 0 { + return "", fmt.Errorf("firstMatch: regex required") + } + pattern, _ := args[0].(string) + re, err := regexp.Compile(pattern) + if err != nil { + return "", fmt.Errorf("firstMatch: %w", err) + } + var texts []string + if len(args) > 1 { + switch t := args[1].(type) { + case string: + texts = []string{t} + case []string: + texts = t + case []any: + for _, e := range t { + texts = append(texts, toString(e)) + } + default: + return "", fmt.Errorf("firstMatch: unsupported text argument") + } + } + for _, t := range texts { + if m := re.FindStringSubmatch(t); m != nil { + if len(m) > 1 { + return m[1], nil + } + return m[0], nil + } + } + return "", nil +} + +func toString(v any) string { + switch t := v.(type) { + case string: + return t + case nil: + return "" + default: + return fmt.Sprintf("%v", t) + } +} + +var frenchMonths = []string{"Janvier", "Février", "Mars", "Avril", "Mai", "Juin", + "Juillet", "Août", "Septembre", "Octobre", "Novembre", "Décembre"} + +// Period is the period(value, unit, shift, layout?) helper (arch.md §5). +func Period(args ...any) (any, error) { + if len(args) < 3 { + return "", fmt.Errorf("period: value, unit and shift required") + } + d, ok := args[0].(time.Time) + if !ok { + return "", fmt.Errorf("period: value must be a native date") + } + unit, _ := args[1].(string) + shift, _ := toInt(args[2]) + var layout string + if len(args) > 3 { + layout, _ = args[3].(string) + } + + y, m := d.Year(), int(d.Month()) + switch unit { + case "month": + total := (y*12 + (m - 1)) + shift + y2, m2 := total/12, total%12 + if m2 < 0 { + m2 += 12 + y2-- + } + month := frenchMonths[m2] + if layout == "" { + layout = "January 2006" + } + first := time.Date(y2, time.Month(m2)+1, 1, 0, 0, 0, 0, time.UTC) + s := first.Format(layout) + return strings.Replace(s, "January", month, 1), nil + case "quarter": + q := (m-1)/3 + shift + for q < 0 { + q += 4 + y-- + } + for q > 3 { + q -= 4 + y++ + } + return fmt.Sprintf("T%d %d", q+1, y), nil + case "half": + h := (m-1)/6 + shift + for h < 0 { + h += 2 + y-- + } + for h > 1 { + h -= 2 + y++ + } + return fmt.Sprintf("S%d %d", h+1, y), nil + case "year": + y += shift + if layout == "" { + layout = "2006" + } + return time.Date(y, 1, 1, 0, 0, 0, 0, time.UTC).Format(layout), nil + default: + return "", fmt.Errorf("period: unknown unit %q (want month|quarter|half|year)", unit) + } +} + +func toInt(v any) (int, bool) { + switch t := v.(type) { + case int: + return t, true + case int64: + return int(t), true + case float64: + return int(t), true + case string: + n, err := strconv.Atoi(t) + return n, err == nil + } + return 0, false +} diff --git a/archivist/internal/config/render.go b/archivist/internal/config/render.go new file mode 100644 index 0000000..0cb2614 --- /dev/null +++ b/archivist/internal/config/render.go @@ -0,0 +1,95 @@ +package config + +import ( + "fmt" + "regexp" + "strconv" + "strings" + "time" + + "github.com/expr-lang/expr" +) + +var placeholderRe = regexp.MustCompile(`\{([^{}]+)\}`) + +type tplPart struct { + literal string + program expr.Node + hasExpr bool +} + +// CompiledTemplate is a naming template with pre-compiled {expr} placeholders. +type CompiledTemplate struct { + Name string + parts []tplPart +} + +// CompileTemplate parses and compiles every placeholder expression (US-02). +func CompileTemplate(name, src string) (*CompiledTemplate, error) { + ct := &CompiledTemplate{Name: name} + last := 0 + for _, loc := range placeholderRe.FindAllStringSubmatchIndex(src, -1) { + if loc[0] > last { + ct.parts = append(ct.parts, tplPart{literal: src[last:loc[0]]}) + } + inner := src[loc[2]:loc[3]] + prog, err := CompileExpr(inner) + if err != nil { + return nil, fmt.Errorf("placeholder {%s}: %w", inner, err) + } + ct.parts = append(ct.parts, tplPart{program: prog, hasExpr: true}) + last = loc[1] + } + if last < len(src) { + ct.parts = append(ct.parts, tplPart{literal: src[last:]}) + } + return ct, nil +} + +// Render evaluates the placeholders in the document context and concatenates +// the result. Unset optional fields render as an empty string; a rendered +// value containing "/" is invalid (it is the S3 key delimiter) (US-15). +func (ct *CompiledTemplate) Render(fields map[string]any, pages []string) (string, error) { + env := ExprEnv(pages, fields) + var b strings.Builder + for _, p := range ct.parts { + if !p.hasExpr { + b.WriteString(p.literal) + continue + } + v, err := expr.Run(p.program, env) + if err != nil { + return "", fmt.Errorf("template render error: %v", err) + } + s, err := renderValue(v) + if err != nil { + return "", err + } + if strings.Contains(s, "/") { + return "", fmt.Errorf("rendered value %q must not contain '/'", s) + } + b.WriteString(s) + } + return b.String(), nil +} + +func renderValue(v any) (string, error) { + switch t := v.(type) { + case nil: + return "", nil + case string: + return t, nil + case time.Time: + return t.Format("2006-01-02"), nil + case float64: + return strconv.FormatFloat(t, 'f', -1, 64), nil + case int: + return strconv.Itoa(t), nil + case int64: + return strconv.FormatInt(t, 10), nil + case bool: + return strconv.FormatBool(t), nil + default: + return "", fmt.Errorf("cannot render value of type %T", v) + } +} diff --git a/archivist/internal/pbapp/hooks.go b/archivist/internal/pbapp/hooks.go new file mode 100644 index 0000000..3592352 --- /dev/null +++ b/archivist/internal/pbapp/hooks.go @@ -0,0 +1,200 @@ +package pbapp + +import ( + "encoding/json" + "fmt" + "log" + "time" + + "github.com/napnap75/archivist/internal/config" + "github.com/napnap75/archivist/internal/storage" + "github.com/pocketbase/pocketbase/core" +) + +// BindHooks registers the request hooks (US-04, US-22, US-23). +func BindHooks(app *core.App, cfg *config.Config, s3 *storage.Client) { + // US-04: document_events is append-only for everyone, superusers included. + app.OnRecordUpdateRequest(eventsName).PreBindFunc(func(e *core.RecordRequestEvent) error { + return errAppendOnly + }) + app.OnRecordDeleteRequest(eventsName).PreBindFunc(func(e *core.RecordRequestEvent) error { + return errAppendOnly + }) + + // US-22 / US-23: corrections and validations through the standard UI. + app.OnRecordUpdateRequest("documents").PreBindFunc(func(e *core.RecordRequestEvent) error { + return handleDocumentUpdate(app, cfg, s3, e) + }) +} + +func handleDocumentUpdate(app *core.App, cfg *config.Config, s3 *storage.Client, e *core.RecordRequestEvent) error { + newRec := e.Record + + // The original state, before the submitted data is persisted. + orig, err := app.FindRecordById("documents", newRec.Id) + if err != nil { + return err + } + oldStatus := orig.GetString("status") + newStatus := newRec.GetString("status") + fieldsChanged := false + for _, f := range cfg.FieldList { + if fmt.Sprintf("%v", orig.Get(f.Name)) != fmt.Sprintf("%v", newRec.Get(f.Name)) { + fieldsChanged = true + break + } + } + + if !fieldsChanged { + // US-23: direct validation, processed -> validated, no field change. + if newStatus == "validated" && oldStatus == "processed" { + before := Snapshot{"status": oldStatus} + after := Snapshot{"status": "validated"} + if err := AppendEvent(app, newRec.Id, "validated", before, after, nil); err != nil { + return err + } + return e.Next() + } + if newStatus != oldStatus { + return fmt.Errorf("refused: the only direct status change is processed -> validated; other changes go through a field correction") + } + return e.Next() + } + + // US-22: a field correction. Whatever the previous status, the document + // becomes validated after a successful re-filing. + fields := FieldsFromRecord(cfg, newRec) + tplName := "default" + if v, ok := fields["template"].(string); ok && v != "" { + tplName = v + } + tpl, ok := cfg.Templates[tplName] + if !ok { + return fmt.Errorf("refused: unknown template %q", tplName) + } + rendered, err := tpl.Render(fields, nil) + if err != nil { + return fmt.Errorf("refused: %v", err) + } + + oldKey := orig.GetString("file_key") + target := "files/" + rendered + finalKey := oldKey + if target != oldKey { + finalKey, err = s3.ResolveConflict(e.Request.Context(), target) + if err != nil { + return err + } + if err := s3.Copy(e.Request.Context(), oldKey, finalKey); err != nil { + return fmt.Errorf("refused: re-filing failed: %v", err) + } + if err := s3.Remove(e.Request.Context(), oldKey); err != nil { + // Roll back the partial copy so the S3 state matches the record. + _ = s3.Remove(e.Request.Context(), finalKey) + return fmt.Errorf("refused: re-filing failed: %v", err) + } + } + newRec.Set("file_key", finalKey) + newRec.Set("status", "validated") + + before := SnapshotFromRecord(cfg, orig, true) + after := SnapshotFromRecord(cfg, newRec, true) + if err := AppendEvent(app, newRec.Id, "corrected", before, after, nil); err != nil { + log.Printf("corrected event for %s: %v", newRec.Id, err) + } + return e.Next() +} + +// FieldsFromRecord reads the declared field values with their native types. +func FieldsFromRecord(cfg *config.Config, record *core.Record) map[string]any { + out := map[string]any{} + for _, f := range cfg.FieldList { + switch f.Type { + case "date": + if !record.GetDateTime(f.Name).IsZero() { + out[f.Name] = record.GetDateTime(f.Name).Time() + } + case "number": + if record.GetFloat(f.Name) != 0 { + out[f.Name] = record.GetFloat(f.Name) + } + case "boolean": + if record.GetBool(f.Name) { + out[f.Name] = true + } + default: + if s := record.GetString(f.Name); s != "" { + out[f.Name] = s + } + } + } + return out +} + +// Snapshot is one before/after state of a document (arch.md §4). +type Snapshot map[string]any + +// SnapshotFromRecord builds a snapshot: file_key always, status when it is +// set, and the recognized declared fields. +func SnapshotFromRecord(cfg *config.Config, record *core.Record, withStatus bool) Snapshot { + s := Snapshot{"file_key": record.GetString("file_key")} + if withStatus { + if st := record.GetString("status"); st != "" { + s["status"] = st + } + } + for name, v := range FieldsFromRecord(cfg, record) { + s[name] = v + } + return s +} + +// Payload is the self-contained event payload (arch.md §4). +type Payload struct { + EventType string `json:"event_type"` + Timestamp string `json:"timestamp"` + Errors []string `json:"errors,omitempty"` + Before Snapshot `json:"before"` + After Snapshot `json:"after"` +} + +// AppendEvent appends one event to document_events. +func AppendEvent(app *core.App, documentId, eventType string, before, after Snapshot, errs []string) error { + coll, err := app.FindCollectionByNameOrId(eventsName) + if err != nil { + return err + } + now := time.Now().UTC() + p := Payload{ + EventType: eventType, + Timestamp: now.Format(time.RFC3339), + Errors: errs, + Before: before, + After: after, + } + data, err := json.Marshal(p) + if err != nil { + return err + } + rec := core.NewRecord(coll) + rec.Set("document", documentId) + rec.Set("event_type", eventType) + rec.Set("payload", string(data)) + rec.Set("timestamp", now) + return app.Save(rec) +} + +// AppendImportedEvent builds the imported event of US-16. +func AppendImportedEvent(app *core.App, documentId, fromKey, toKey string, valid bool, fields map[string]any, errs []string) error { + before := Snapshot{"file_key": fromKey} + after := Snapshot{"file_key": toKey} + if valid { + after["status"] = "processed" + } else { + after["status"] = "pending" + } + for k, v := range fields { + after[k] = v + } + return AppendEvent(app, documentId, "imported", before, after, errs) +} diff --git a/archivist/internal/pbapp/setup.go b/archivist/internal/pbapp/setup.go new file mode 100644 index 0000000..469e766 --- /dev/null +++ b/archivist/internal/pbapp/setup.go @@ -0,0 +1,146 @@ +// Package pbapp sets up the PocketBase collections, superuser account and +// request hooks (US-03, US-04, US-22, US-23). +package pbapp + +import ( + "errors" + "fmt" + "log" + + "github.com/napnap75/archivist/internal/config" + "github.com/pocketbase/pocketbase/core" +) + +const eventsName = "document_events" + +// Setup ensures the collections and the superuser account (US-03, US-04, US-01). +func Setup(app *core.App, adminEmail, adminPassword string, cfg *config.Config) error { + if err := ensureDocuments(app, cfg); err != nil { + return err + } + if err := ensureEvents(app); err != nil { + return err + } + if err := ensureSuperuser(app, adminEmail, adminPassword); err != nil { + return err + } + config.SetKnownFields(cfg.FieldList) + return nil +} + +func ensureDocuments(app *core.App, cfg *config.Config) error { + coll, err := app.FindCollectionByNameOrId("documents") + if err != nil { + // First start: create the collection with its fixed columns (US-03). + coll = core.NewBaseCollection("documents") + } + + // Fixed columns (re)declared on every start, ordering preserved. + coll.Fields.RemoveByName("sha256") + coll.Fields.RemoveByName("file_key") + coll.Fields.RemoveByName("pages") + coll.Fields.RemoveByName("status") + coll.Fields.Add( + &core.TextField{Name: "sha256"}, + &core.TextField{Name: "file_key"}, + &core.JSONField{Name: "pages"}, + &core.SelectField{Name: "status", MaxSelect: 1, Values: []string{"pending", "processed", "validated"}}, + ) + + // One column per declared field, mapped from the field type (US-03). + for _, f := range cfg.FieldList { + var newField core.Field + switch f.Type { + case "text": + newField = &core.TextField{Name: f.Name} + case "number": + newField = &core.NumberField{Name: f.Name} + case "date": + newField = &core.DateField{Name: f.Name} + case "boolean": + newField = &core.BoolField{Name: f.Name} + case "enum": + // The vocabulary is an application notion (dashboard dropdown with + // free entry), not a schema constraint: empty allowed values. + newField = &core.SelectField{Name: f.Name, MaxSelect: 1} + } + if existing := coll.Fields.GetByName(f.Name); existing != nil { + if existing.Type() != newField.Type() { + return fmt.Errorf("field %s changed type from %s to %s: refusing to start", + f.Name, existing.Type(), newField.Type()) + } + continue + } + coll.Fields.Add(newField) + } + + // Fields removed from the config are removed from the schema; values are + // lost by design (they remain in the event snapshots). + for _, f := range coll.Fields { + if f.GetSystem() { + continue + } + if _, declared := cfg.Fields[f.Name]; !declared && !isFixed(f.Name) { + log.Printf("documents: removing column %q (removed from config)", f.Name) + coll.Fields.RemoveByName(f.Name) + } + } + + if err := app.Save(coll); err != nil { + return fmt.Errorf("save documents collection: %w", err) + } + if _, err := app.DB().NewQuery( + "CREATE UNIQUE INDEX IF NOT EXISTS idx_documents_sha256 ON documents (sha256)").Execute(); err != nil { + return fmt.Errorf("sha256 unique index: %w", err) + } + return nil +} + +func isFixed(name string) bool { + for _, f := range []string{"sha256", "file_key", "pages", "status"} { + if f == name { + return true + } + } + return false +} + +func ensureEvents(app *core.App) error { + if _, err := app.FindCollectionByNameOrId(eventsName); err == nil { + return nil + } + docColl, err := app.FindCollectionByNameOrId("documents") + if err != nil { + return err + } + coll := core.NewBaseCollection(eventsName) + coll.Fields.Add( + &core.RelationField{Name: "document", CollectionId: docColl.Id, MaxSelect: 1, CascadeDelete: true}, + &core.SelectField{Name: "event_type", MaxSelect: 1, Values: []string{"imported", "corrected", "validated"}}, + &core.JSONField{Name: "payload"}, + &core.DateField{Name: "timestamp"}, + ) + // Append-only (US-04): API rules locked to superusers are not enough, + // the standard admin UI uses a superuser; the request hooks in + // BindHooks reject updates and deletes for everyone. + if err := app.Save(coll); err != nil { + return fmt.Errorf("save %s collection: %w", eventsName, err) + } + return nil +} + +func ensureSuperuser(app *core.App, email, password string) error { + superusers, err := app.FindCollectionByNameOrId(core.CollectionNameSuperusers) + if err != nil { + return err + } + rec, err := app.FindFirstRecordByFilter(superusers.Name, "email = {:e}", map[string]any{"e": email}) + if err != nil || rec == nil { + rec = core.NewRecord(superusers) + rec.SetEmail(email) + } + rec.SetPassword(password) + return app.Save(rec) +} + +var errAppendOnly = errors.New("document_events is append-only: update and delete are refused") diff --git a/archivist/internal/pipeline/pipeline.go b/archivist/internal/pipeline/pipeline.go new file mode 100644 index 0000000..3c127b6 --- /dev/null +++ b/archivist/internal/pipeline/pipeline.go @@ -0,0 +1,218 @@ +// Package pipeline implements US-10 to US-17: polling, download, +// extraction, classification, recording and filing. +package pipeline + +import ( + "context" + "encoding/json" + "fmt" + "log" + "strings" + + "github.com/ledongthuc/pdf" + "github.com/napnap75/archivist/internal/config" + "github.com/napnap75/archivist/internal/pbapp" + "github.com/napnap75/archivist/internal/storage" + "github.com/pocketbase/pocketbase/core" + "github.com/pocketbase/pocketbase/tools/dbx" +) + +// pollState remembers size and ETag between two passes (US-10 write-completion wait). +type pollState struct { + size int64 + etag string +} + +var stable = map[string]pollState{} + +func RunPoll(app *core.App, cfg *config.Config, s3 *storage.Client) { + ctx := context.Background() + + // US-11: recover interrupted filings first. + recoverStuck(app, s3, ctx) + + objects, err := s3.ListInbox(ctx) + if err != nil { + log.Printf("poll: list inbox: %v", err) + return + } + + for _, o := range objects { + // US-10: an object referenced by a record is not a candidate. + linked, err := app.FindFirstRecordByFilter("documents", "file_key = {:k}", dbx.Params{"k": o.Key}) + if err == nil && linked != nil { + continue + } + prev, seen := stable[o.Key] + stable[o.Key] = pollState{size: o.Size, etag: o.ETag} + if !seen || prev.size != o.Size || prev.etag != o.ETag { + continue // first sighting, or still uploading: wait for stability + } + delete(stable, o.Key) + if err := processOne(app, cfg, s3, ctx, o.Key); err != nil { + log.Printf("poll: process %s: %v", o.Key, err) + } + } +} + +func processOne(app *core.App, cfg *config.Config, s3 *storage.Client, ctx context.Context, key string) error { + // US-12: download in memory, sha256 on the fly. + data, hash, err := s3.Get(ctx, key) + if err != nil { + return fmt.Errorf("download: %w", err) + } + + // US-12: duplicate rejection. + dup, err := app.FindFirstRecordByFilter("documents", "sha256 = {:h}", dbx.Params{"h": hash}) + if err == nil && dup != nil { + if err := s3.Remove(ctx, key); err != nil { + return fmt.Errorf("duplicate cleanup: %w", err) + } + log.Printf("duplicate %s (sha256 %s) removed from inbox, original is %s", key, hash, dup.GetString("file_key")) + return nil + } + + // US-13: page-by-page extraction. + pages, extractErr := extractPages(data) + var errs []string + if extractErr != nil { + errs = append(errs, "unreadable PDF: "+extractErr.Error()) + pages = nil + } + + // US-14 / US-15: classification and verdict. + fields, rendered, evalErrs := cfg.Evaluate(pages) + errs = append(errs, evalErrs...) + valid := len(errs) == 0 && rendered != "" + + // US-16: the record is created before any physical move. + coll, err := app.FindCollectionByNameOrId("documents") + if err != nil { + return err + } + record := core.NewRecord(coll) + record.Set("sha256", hash) + record.Set("file_key", key) + pagesJSON, _ := json.Marshal(pages) + record.Set("pages", string(pagesJSON)) + for name, v := range fields { + record.Set(name, v) + } + if valid { + record.Set("status", "processed") + } else { + record.Set("status", "pending") + } + if err := app.Save(record); err != nil { + return fmt.Errorf("save record: %w", err) + } + + // US-17: filing. + newKey, err := fileDocument(s3, ctx, record, rendered) + if err != nil { + return err + } + if err := app.Save(record); err != nil { + return fmt.Errorf("update record: %w", err) + } + + // US-16: the imported event, appended once the filing is done. + if err := pbapp.AppendImportedEvent(app, record.Id, key, newKey, valid, fields, errs); err != nil { + log.Printf("imported event for %s: %v", key, err) + } + log.Printf("imported %s -> %s (status %s)", key, newKey, record.GetString("status")) + return nil +} + +// fileDocument moves the object to files/ (valid) or pending/ (invalid), +// updates file_key, and returns the final key. The record save is left to the +// caller so the state stays consistent on recovery. +func fileDocument(s3 *storage.Client, ctx context.Context, record *core.Record, rendered string) (string, error) { + valid := record.GetString("status") == "processed" + var target string + if valid { + // The rendered template is the full object path under files/. + target = "files/" + rendered + } else { + target = "pending/" + baseName(record.GetString("file_key")) + } + finalKey, err := s3.ResolveConflict(ctx, target) + if err != nil { + return "", err + } + if err := s3.Copy(ctx, record.GetString("file_key"), finalKey); err != nil { + return "", fmt.Errorf("copy: %w", err) + } + if !s3.Exists(ctx, finalKey) { + return "", fmt.Errorf("target %s not verified after copy", finalKey) + } + if err := s3.Remove(ctx, record.GetString("file_key")); err != nil { + return "", fmt.Errorf("remove source: %w", err) + } + record.Set("file_key", finalKey) + return finalKey, nil +} + +// recoverStuck completes interrupted filings: records whose file_key is still +// under inbox/ (US-11). +func recoverStuck(app *core.App, s3 *storage.Client, ctx context.Context) { + stuck, err := app.FindRecordsByFilter("documents", "file_key ~ {:p}", "", 100, 0, dbx.Params{"p": "inbox/"}) + if err != nil { + log.Printf("recovery: %v", err) + return + } + for _, record := range stuck { + rendered := "" + if record.GetString("status") == "processed" { + // Re-render the template from the stored fields. + tplName := record.GetString("template") + if tplName == "" { + tplName = "default" + } + _loop: + _ = tplName + } + newKey, err := fileDocument(s3, ctx, record, rendered) + if err != nil { + log.Printf("recovery: record %s (%s): %v", record.Id, record.GetString("file_key"), err) + continue + } + if err := app.Save(record); err != nil { + log.Printf("recovery: save record %s: %v", record.Id, err) + continue + } + log.Printf("recovery: record %s completed filing to %s", record.Id, newKey) + } +} + +func baseName(key string) string { + if i := strings.LastIndex(key, "/"); i >= 0 { + return key[i+1:] + } + return key +} + +// extractPages extracts the PDF text page by page including headers and +// footers (US-13). +func extractPages(data []byte) ([]string, error) { + r, err := pdf.NewReader(bytesReader(data), int64(len(data))) + if err != nil { + return nil, err + } + var pages []string + for i := 1; i <= r.NumPage(); i++ { + p := r.Page(i) + if p.V.IsNull() { + return nil, fmt.Errorf("page %d is null", i) + } + pages = append(pages, p.GetPlainText(nil)) + } + if len(pages) == 0 { + return nil, fmt.Errorf("no page extracted") + } + return pages, nil +} + +func bytesReader(data []byte) *strings.Reader { + return strings.NewReader(string(data)) +} diff --git a/archivist/internal/storage/s3.go b/archivist/internal/storage/s3.go new file mode 100644 index 0000000..539a2fe --- /dev/null +++ b/archivist/internal/storage/s3.go @@ -0,0 +1,123 @@ +// Package storage wraps the MinIO S3 client (US-01, US-10..US-17). +package storage + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "io" + "regexp" + "strconv" + "strings" + + "github.com/minio/minio-go/v7" + "github.com/minio/minio-go/v7/pkg/credentials" +) + +type Client struct { + mc *minio.Client + Bucket string +} + +func New(endpoint, accessKey, secretKey, bucket string, useSSL bool) (*Client, error) { + mc, err := minio.New(endpoint, &minio.Options{ + Creds: credentials.NewStaticV4(accessKey, secretKey, ""), + Secure: useSSL, + }) + if err != nil { + return nil, err + } + c := &Client{mc: mc, Bucket: bucket} + + ctx := context.Background() + exists, err := mc.BucketExists(ctx, bucket) + if err != nil { + return nil, fmt.Errorf("bucket check: %w", err) + } + if !exists { + return nil, fmt.Errorf("bucket %q does not exist", bucket) + } + // US-01: the bucket must be versioned (delete markers keep data recoverable). + vc, err := mc.GetBucketVersioning(ctx, bucket) + if err != nil { + return nil, fmt.Errorf("versioning check: %w", err) + } + if vc.Status != "Enabled" { + return nil, fmt.Errorf("bucket %q is not versioned (status %q)", bucket, vc.Status) + } + return c, nil +} + +// InboxObject is one object listed under inbox/. +type InboxObject struct { + Key string + Size int64 + ETag string +} + +// ListInbox lists the objects under the inbox/ prefix (US-10). +func (c *Client) ListInbox(ctx context.Context) ([]InboxObject, error) { + var out []InboxObject + for obj := range c.mc.ListObjects(ctx, c.Bucket, minio.ListObjectsOptions{Prefix: "inbox/", Recursive: true}) { + if obj.Err != nil { + return nil, obj.Err + } + out = append(out, InboxObject{Key: obj.Key, Size: obj.Size, ETag: obj.ETag}) + } + return out, nil +} + +// Get downloads the object into memory and computes its sha256 on the fly +// (US-12: streamed in memory, no disk writes). +func (c *Client) Get(ctx context.Context, key string) ([]byte, string, error) { + obj, err := c.mc.GetObject(ctx, c.Bucket, key, minio.GetObjectOptions{}) + if err != nil { + return nil, "", err + } + defer obj.Close() + data, err := io.ReadAll(obj) + if err != nil { + return nil, "", err + } + sum := sha256.Sum256(data) + return data, hex.EncodeToString(sum[:]), nil +} + +func (c *Client) Exists(ctx context.Context, key string) bool { + _, err := c.mc.StatObject(ctx, c.Bucket, key, minio.StatObjectOptions{}) + return err == nil +} + +func (c *Client) Copy(ctx context.Context, srcKey, dstKey string) error { + _, err := c.mc.CopyObject(ctx, + minio.CopyDestOptions{Bucket: c.Bucket, Object: dstKey}, + minio.CopySourceOptions{Bucket: c.Bucket, Object: srcKey}) + return err +} + +// Remove deletes the object; with a versioned bucket this creates a delete +// marker and the data remains recoverable. +func (c *Client) Remove(ctx context.Context, key string) error { + return c.mc.RemoveObject(ctx, c.Bucket, key, minio.RemoveObjectOptions{}) +} + +var extRe = regexp.MustCompile(`(\.[^./]+)$`) + +// ResolveConflict returns desired when no object exists under that key, +// otherwise the name with an incremental numeric suffix before the extension +// (US-15, US-17): "name.pdf" -> "name (1).pdf", "name (2).pdf", ... +func (c *Client) ResolveConflict(ctx context.Context, desired string) (string, error) { + if !c.Exists(ctx, desired) { + return desired, nil + } + ext := extRe.FindString(desired) + base := strings.TrimSuffix(desired, ext) + for n := 1; n < 1000; n++ { + candidate := base + " (" + strconv.Itoa(n) + ")" + ext + if !c.Exists(ctx, candidate) { + return candidate, nil + } + } + return "", fmt.Errorf("cannot resolve conflict on %q", desired) +} diff --git a/archivist/main.go b/archivist/main.go new file mode 100644 index 0000000..c2c59c5 --- /dev/null +++ b/archivist/main.go @@ -0,0 +1,104 @@ +package main + +import ( + "log" + "os" + + "github.com/napnap75/archivist/internal/config" + "github.com/napnap75/archivist/internal/pbapp" + "github.com/napnap75/archivist/internal/pipeline" + "github.com/napnap75/archivist/internal/storage" + "github.com/pocketbase/pocketbase" +) + +// EnvConfig is the technical configuration, from environment variables (US-01). +type EnvConfig struct { + S3Endpoint string + S3AccessKey string + S3SecretKey string + S3Bucket string + S3UseSSL bool + AdminEmail string + AdminPassword string + PollCron string + HTTPAddr string + ConfigPath string +} + +func env(key, def string) string { + if v := os.Getenv(key); v != "" { + return v + } + return def +} + +// LoadEnvConfig reads and validates the mandatory environment variables (US-01). +func LoadEnvConfig() (*EnvConfig, error) { + c := &EnvConfig{ + S3Endpoint: os.Getenv("ARCHIVIST_S3_ENDPOINT"), + S3AccessKey: os.Getenv("ARCHIVIST_S3_ACCESS_KEY"), + S3SecretKey: os.Getenv("ARCHIVIST_S3_SECRET_KEY"), + S3Bucket: os.Getenv("ARCHIVIST_S3_BUCKET"), + AdminEmail: os.Getenv("ARCHIVIST_ADMIN_EMAIL"), + AdminPassword: os.Getenv("ARCHIVIST_ADMIN_PASSWORD"), + } + for k, v := range map[string]string{ + "ARCHIVIST_S3_ENDPOINT": c.S3Endpoint, + "ARCHIVIST_S3_ACCESS_KEY": c.S3AccessKey, + "ARCHIVIST_S3_SECRET_KEY": c.S3SecretKey, + "ARCHIVIST_S3_BUCKET": c.S3Bucket, + "ARCHIVIST_ADMIN_EMAIL": c.AdminEmail, + "ARCHIVIST_ADMIN_PASSWORD": c.AdminPassword, + } { + if v == "" { + log.Fatalf("missing mandatory environment variable %s", k) + } + } + c.S3UseSSL = env("ARCHIVIST_S3_USE_SSL", "true") != "false" + c.PollCron = env("ARCHIVIST_POLL_INTERVAL", "* * * * *") + c.HTTPAddr = env("ARCHIVIST_HTTP_ADDR", "0.0.0.0:8090") + c.ConfigPath = env("ARCHIVIST_CONFIG", "config.jsonc") + return c, nil +} + +func main() { + envCfg, err := LoadEnvConfig() + if err != nil { + log.Fatal(err) + } + + // US-02: load and validate the functional configuration. Full rejection, + // the process does not continue with a partial configuration (US-05). + cfg, err := config.Load(envCfg.ConfigPath) + if err != nil { + log.Fatalf("invalid functional configuration: %v", err) + } + log.Printf("configuration loaded: %d fields, %d top-level rules, %d templates", + len(cfg.Fields), len(cfg.Rules), len(cfg.Templates)) + + // US-01: verify S3 reachability and versioning before anything else. + s3, err := storage.New(envCfg.S3Endpoint, envCfg.S3AccessKey, envCfg.S3SecretKey, envCfg.S3Bucket, envCfg.S3UseSSL) + if err != nil { + log.Fatalf("S3 client: %v", err) + } + + app := pocketbase.New() + + // US-03 / US-04 / US-01: collections migration and superuser account. + if err := pbapp.Setup(app, envCfg.AdminEmail, envCfg.AdminPassword, cfg); err != nil { + log.Fatalf("pocketbase setup: %v", err) + } + + // US-22 / US-23 / US-04: request hooks on documents and document_events. + pbapp.BindHooks(app, cfg, s3) + + // US-10..US-17: the poller, on the configured cron. + app.CronAdd(envCfg.PollCron, "archivist-poller", func() { + pipeline.RunPoll(app, cfg, s3) + }) + + app.RootCmd.PersistentFlags().Set("http", envCfg.HTTPAddr) + if err := app.Start(); err != nil { + log.Fatal(err) + } +}