feat: index the text of attached PDFs

- jz_bm_material_file_texts caches the pdftotext output per file
- the River job jz.bm.extract_material_text extracts and resyncs
- the management multipart attach queues it inside its transaction
- file_texts in the document and a file snippet with its name

classes/material_index.go

@@ -13,8 +13,11 @@ import (
 	"git.golem15.com/golem15/summercms/modules/backpack"
 	"git.golem15.com/golem15/summercms/modules/beachcomber"
 	"git.golem15.com/golem15/summercms/modules/beachcomber/typesense"
+	"git.golem15.com/golem15/summercms/modules/conga"
 	"git.golem15.com/golem15/summercms/modules/diary"
 	"git.golem15.com/golem15/summercms/modules/lagoon"
+	"git.golem15.com/golem15/summercms/modules/pact"
+	"git.golem15.com/jakub/sm-bm-plugin/jobs"
 	"git.golem15.com/jakub/sm-bm-plugin/models"
 	"gorm.io/gorm"
 )
@@ -92,6 +95,9 @@ type MaterialIndexer struct {
 	Service func() *beachcomber.Service
 	// Log is where failures are logged; nil means slog.Default.
 	Log *slog.Logger
+	// Enqueue inserts a background job, inside tx when tx is in a
+	// transaction. Nil means jobs are not queued.
+	Enqueue func(ctx context.Context, tx *gorm.DB, args pact.JobArgs) error
 }
 
 // NewMaterialIndexer returns the indexer of app. The service is resolved on
@@ -109,6 +115,27 @@ func NewMaterialIndexer(app *backpack.App) *MaterialIndexer {
 			return svc
 		},
 		Log: diary.For("jz.bm"),
+		Enqueue: func(ctx context.Context, tx *gorm.DB, args pact.JobArgs) error {
+			m, err := conga.From(app)
+			if err != nil {
+				return err
+			}
+			return m.Enqueue(ctx, tx, args, conga.EnqueueOpts{})
+		},
+	}
+}
+
+// QueueExtract enqueues the file text job of material id inside tx, so the
+// job exists only when the attach commits. The job extracts the text of the
+// material's new PDFs and then syncs its document. A failure is logged and
+// not returned: an attach never fails because of the job, and
+// bm:materials-reindex backfills the text.
+func (ix *MaterialIndexer) QueueExtract(ctx context.Context, tx *gorm.DB, id uint) {
+	if ix == nil || ix.Enqueue == nil || id == 0 {
+		return
+	}
+	if err := ix.Enqueue(ctx, tx, jobs.ExtractMaterialTextArgs{MaterialID: id}); err != nil {
+		ix.log().Warn("jz.bm: could not queue the file text job", "material_id", id, "err", err)
 	}
 }
 

classes/material_texts.go

@@ -0,0 +1,76 @@
+package classes
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strconv"
+	"time"
+
+	"git.golem15.com/golem15/summercms/modules/lagoon/attach"
+	"git.golem15.com/jakub/sm-bm-plugin/models"
+	"gocloud.dev/blob"
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+)
+
+// EnsureFileTexts extracts the text of every current public PDF of material
+// materialID that has no text row yet, and stores one row per file (an
+// empty text for a scan or a broken PDF, so it is not retried). It returns
+// how many rows it added. ErrPDFToolMissing stops the loop and is returned
+// as is, so callers can match it; a blob, extraction or insert error is
+// returned wrapped with the file id. It runs only in the extraction job and
+// bm:materials-reindex, never in a request.
+func EnsureFileTexts(ctx context.Context, db *gorm.DB, bucket *blob.Bucket, pdf PDFText, materialID uint) (int, error) {
+	if db == nil {
+		return 0, fmt.Errorf("jz.bm: no database for the file texts of material %d", materialID)
+	}
+	if bucket == nil {
+		return 0, fmt.Errorf("jz.bm: no uploads bucket for the file texts of material %d", materialID)
+	}
+	db = db.WithContext(ctx)
+	// The conditions of the API's materialFiles, which classes cannot import.
+	var files []attach.File
+	err := db.Model(&attach.File{}).
+		Select("system_files.*").
+		Joins("LEFT JOIN jz_bm_material_file_texts t ON t.file_id = system_files.id").
+		Where("system_files.attachment_type = ? AND system_files.attachment_id = ? AND system_files.field = ?",
+			models.Material{}.MorphName(), strconv.FormatUint(uint64(materialID), 10), models.MaterialFilesField).
+		Where("system_files.is_public").
+		Where("(system_files.content_type = 'application/pdf' OR lower(system_files.file_name) LIKE '%.pdf')").
+		Where("t.file_id IS NULL").
+		Order("system_files.sort_order NULLS LAST, system_files.id").
+		Find(&files).Error
+	if err != nil {
+		return 0, err
+	}
+	added := 0
+	for _, f := range files {
+		text, truncated, err := extractBlob(ctx, bucket, pdf, f)
+		if errors.Is(err, ErrPDFToolMissing) {
+			return added, err
+		}
+		if err != nil {
+			return added, fmt.Errorf("jz.bm: file %d: %w", f.ID, err)
+		}
+		row := models.MaterialFileText{FileID: f.ID, MaterialID: materialID, Text: text, Truncated: truncated, ExtractedAt: time.Now().UTC()}
+		res := db.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "file_id"}}, DoNothing: true}).Create(&row)
+		if res.Error != nil {
+			return added, fmt.Errorf("jz.bm: file %d: %w", f.ID, res.Error)
+		}
+		if res.RowsAffected > 0 {
+			added++
+		}
+	}
+	return added, nil
+}
+
+// extractBlob runs pdf on the blob of f.
+func extractBlob(ctx context.Context, bucket *blob.Bucket, pdf PDFText, f attach.File) (string, bool, error) {
+	rd, err := bucket.NewReader(ctx, attach.BlobKey(f.DiskName), nil)
+	if err != nil {
+		return "", false, err
+	}
+	defer rd.Close()
+	return pdf.Extract(ctx, rd)
+}

classes/pdftext.go

@@ -0,0 +1,124 @@
+package classes
+
+import (
+	"context"
+	"errors"
+	"io"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"time"
+)
+
+// ErrPDFToolMissing is returned by Extract when the pdftotext binary is not
+// installed (poppler-utils). Nothing is stored then, so a later run
+// retries.
+var ErrPDFToolMissing = errors.New("jz.bm: pdftotext is not installed")
+
+// PDFText runs pdftotext (poppler-utils) on one PDF. The PDF comes from a
+// material manager, so it is untrusted input to a native tool: it runs
+// without a shell, from a 0600 temp file in a 0700 directory that is
+// removed afterwards, for at most Timeout and with at most MaxBytes of text
+// read. It runs only in the extraction job and the reindex command, never in
+// a request.
+type PDFText struct {
+	// Bin is the pdftotext binary, looked up in PATH.
+	Bin string
+	// Timeout bounds one run.
+	Timeout time.Duration
+	// MaxBytes is the most text kept per file.
+	MaxBytes int
+	// TempDir is where the PDF is copied for the tool.
+	TempDir string
+}
+
+// DefaultPDFText is pdftotext with a 60 s timeout and 1 MiB of text per
+// file, with its temp files under storage/app/tmp in the app's working
+// directory (beside the uploads bucket, not a tmpfs /tmp).
+func DefaultPDFText() PDFText {
+	return PDFText{
+		Bin:      "pdftotext",
+		Timeout:  60 * time.Second,
+		MaxBytes: 1 << 20,
+		TempDir:  filepath.Join("storage", "app", "tmp"),
+	}
+}
+
+// Extract returns the text of the PDF read from r. A PDF with no text layer
+// (a scan) or a broken one makes pdftotext fail; that is not an error: the
+// text read so far, usually "", comes back with a nil error. truncated is
+// set when the text went over MaxBytes or the run timed out. A missing
+// binary is ErrPDFToolMissing; a cancelled ctx is its error.
+func (p PDFText) Extract(ctx context.Context, r io.Reader) (string, bool, error) {
+	bin := p.Bin
+	if bin == "" {
+		bin = "pdftotext"
+	}
+	dir := p.TempDir
+	if dir == "" {
+		dir = os.TempDir()
+	}
+	if err := os.MkdirAll(dir, 0o700); err != nil {
+		return "", false, err
+	}
+	tmp, err := os.CreateTemp(dir, "jz-bm-pdftext-*.pdf")
+	if err != nil {
+		return "", false, err
+	}
+	path := tmp.Name()
+	defer os.Remove(path)
+	if _, err := io.Copy(tmp, r); err != nil {
+		tmp.Close()
+		return "", false, err
+	}
+	if err := tmp.Close(); err != nil {
+		return "", false, err
+	}
+
+	runCtx := ctx
+	if p.Timeout > 0 {
+		var cancel context.CancelFunc
+		runCtx, cancel = context.WithTimeout(ctx, p.Timeout)
+		defer cancel()
+	}
+	cmd := exec.CommandContext(runCtx, bin, "-q", "-enc", "UTF-8", path, "-")
+	// No stdin (the PDF is a path) and stderr is discarded.
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		return "", false, err
+	}
+	if err := cmd.Start(); err != nil {
+		if errors.Is(err, exec.ErrNotFound) {
+			return "", false, ErrPDFToolMissing
+		}
+		return "", false, err
+	}
+	limit := p.MaxBytes
+	if limit <= 0 {
+		limit = 1 << 20
+	}
+	out, _ := io.ReadAll(io.LimitReader(stdout, int64(limit)+1))
+	truncated := len(out) > limit
+	if truncated {
+		out = out[:limit]
+		// The rest is not wanted; stop the tool instead of draining it.
+		_ = cmd.Process.Kill()
+	}
+	_ = cmd.Wait()
+	if ctx.Err() != nil {
+		return "", false, ctx.Err()
+	}
+	if runCtx.Err() != nil {
+		truncated = true
+	}
+	return CleanExtractedText(string(out)), truncated, nil
+}
+
+// CleanExtractedText removes NUL bytes, which a Postgres TEXT column
+// refuses, and invalid UTF-8 (such as a sequence cut at the size cap) from
+// text a tool extracted.
+func CleanExtractedText(s string) string {
+	s = strings.ReplaceAll(s, "\x00", "")
+	return strings.ToValidUTF8(s, "")
+}

classes/pdftext_test.go

@@ -0,0 +1,89 @@
+package classes_test
+
+import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"errors"
+	"os"
+	"os/exec"
+	"strings"
+	"testing"
+	"time"
+	"unicode/utf8"
+
+	"git.golem15.com/jakub/sm-bm-plugin/classes"
+	"git.golem15.com/jakub/sm-bm-plugin/internal/pdftest"
+)
+
+func TestPDFTextExtract(t *testing.T) {
+	if _, err := exec.LookPath("pdftotext"); err != nil {
+		t.Skip("pdftotext is not installed (poppler-utils)")
+	}
+	ctx := context.Background()
+	dir := t.TempDir()
+	pdf := classes.DefaultPDFText()
+	pdf.TempDir = dir
+	if pdf.Bin != "pdftotext" || pdf.Timeout != 60*time.Second || pdf.MaxBytes != 1<<20 {
+		t.Errorf("DefaultPDFText = %+v, want pdftotext, 60 s, 1 MiB", pdf)
+	}
+	doc := pdftest.OnePage("Globalizacja i demokracja")
+
+	text, truncated, err := pdf.Extract(ctx, bytes.NewReader(doc))
+	if err != nil {
+		t.Fatalf("Extract: %v", err)
+	}
+	if !strings.Contains(text, "Globalizacja i demokracja") || truncated {
+		t.Errorf("Extract = %q truncated %v, want the phrase, not truncated", text, truncated)
+	}
+
+	small := pdf
+	small.MaxBytes = 5
+	text, truncated, err = small.Extract(ctx, bytes.NewReader(doc))
+	if err != nil || !truncated || len(text) > 5 {
+		t.Errorf("MaxBytes 5: %q truncated %v err %v, want at most 5 bytes and truncated", text, truncated, err)
+	}
+
+	junk := make([]byte, 4096)
+	_, _ = rand.Read(junk)
+	text, _, err = pdf.Extract(ctx, bytes.NewReader(junk))
+	if err != nil || text != "" {
+		t.Errorf("random bytes: %q err %v, want empty text and no error", text, err)
+	}
+
+	missing := pdf
+	missing.Bin = "jz-bm-no-such-pdftotext"
+	if _, _, err := missing.Extract(ctx, bytes.NewReader(doc)); !errors.Is(err, classes.ErrPDFToolMissing) {
+		t.Errorf("missing binary: err %v, want ErrPDFToolMissing", err)
+	}
+
+	entries, err := os.ReadDir(dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(entries) != 0 {
+		t.Errorf("the temp dir holds %d files after extraction, want none", len(entries))
+	}
+	// A temp dir Extract creates itself is private to the app user.
+	sub := pdf
+	sub.TempDir = dir + "/sub"
+	if _, _, err := sub.Extract(ctx, bytes.NewReader(doc)); err != nil {
+		t.Fatal(err)
+	}
+	if info, err := os.Stat(sub.TempDir); err != nil || info.Mode().Perm() != 0o700 {
+		t.Errorf("created temp dir: %v %v, want mode 0700", info, err)
+	}
+
+	for _, c := range []struct{ in, want string }{
+		{"plain tekst", "plain tekst"},
+		{"a\x00b\x00", "ab"},
+		{"zażółć\xff\xfe gęślą", "zażółć gęślą"},
+		{"cut \xc5", "cut "},
+		{"", ""},
+	} {
+		got := classes.CleanExtractedText(c.in)
+		if got != c.want || !utf8.ValidString(got) {
+			t.Errorf("CleanExtractedText(%q) = %q, want %q", c.in, got, c.want)
+		}
+	}
+}