feat: extract file text on every attach path and backfill on reindex
- inline and signed uploads queue the text job in their transaction - the admin form queues it in the save transaction - detach resyncs the material (its text row goes by cascade) - bm:materials-reindex backfills missing PDF texts before each sync
This commit is contained in:
@@ -196,6 +196,8 @@ Newest publication first. `meeting` keeps the materials assigned to that zjazd (
|
||||
"segments": [{"text": "proces ", "hit": false}, {"text": "globalizacji", "hit": true}, {"text": " zmienia państwa", "hit": false}]}}
|
||||
```
|
||||
|
||||
A row whose match is only in an attached PDF carries a file snippet instead: `"source": "file"`, `file_name` the name of that file (null when the engine's array index does not resolve to a current file) and `lead` and `trail` both true. A body snippet is preferred when both exist.
|
||||
|
||||
`segments` are the engine's plain text, verbatim, never markup: render each `text` as text and mark the `hit` ones. `lead` and `trail` say that text was cut before or after the excerpt. The key is absent when there is no snippet, and always without `q`, so the list without `q` is byte for byte as before. When the search engine is not configured or does not answer, a request with `q` is a 503 `{"error": true, "message": "Search is unavailable."}`; the list without `q` keeps working and never calls the engine.
|
||||
|
||||
### Private materials
|
||||
@@ -339,7 +341,7 @@ The file tools return metadata and links, never file contents, and their descrip
|
||||
|
||||
- Links: the middleware `jz.bm.mcp-links` (on the token twins and the management API) prefixes every JSON string under a key named `url` or ending in `_url` that starts with a single `/` with `app.url`, else `jz.bm.public_url`, else the request's scheme and Host, so file and syllabus links are ready to follow.
|
||||
- Material bodies (possible transcript text): `bm_material`, `bm_manage_list_materials` and `bm_manage_show_material` leave `body` out unless `include_body=true`, and mark an omitted non-empty body with `body_omitted: true`; the material write answers never echo it. Material tools return metadata and links, never whole file contents.
|
||||
- Search: `bm_materials` and `bm_manage_list_materials` take `q` as the HTTP lists do. With `q` the results come by relevance and a material may carry `snippet` as a plain string, the matched passage with `…` where text was cut; it can come from the body or, once file text is indexed, from an attached file.
|
||||
- Search: `bm_materials` and `bm_manage_list_materials` take `q` as the HTTP lists do. With `q` the results come by relevance and a material may carry `snippet` as a plain string, the matched passage with `…` where text was cut; it can come from the body or from an attached PDF.
|
||||
- Throttling: `jz.bm.manage` keys an MCP call by its user (`jz.bm.manage.user|<id>`) before the bearer and IP keys.
|
||||
|
||||
## Student app
|
||||
@@ -378,6 +380,8 @@ The index holds every material, drafts and private ones included, and no visibil
|
||||
|
||||
The beachcomber service is app-wide, and so is its Gate: jz.bm installs one that is on only for contexts it marks (`classes.MarkMaterialSearch`), so `golem15.chatchannels` messages and DMs are never indexed. Every material write syncs explicitly after it commits: the admin form (create, update, delete), `POST`, `PUT` and `DELETE /api/v1/bm/materials`. A failed sync is logged with the material id and never fails the write.
|
||||
|
||||
Attached PDFs are searchable too. After any attach (the admin form, `POST /api/v1/bm/materials/{id}/files`, `/files/inline` and the signed upload, so also the MCP attach and upload tools) the background job `jz.bm.extract_material_text` (River, through `conga`; enqueued inside the attach transaction, three attempts of at most two minutes) runs `pdftotext` from poppler-utils on each new public PDF of the material and caches the text in `jz_bm_material_file_texts`, then resyncs the material, whose document carries the texts as `file_texts`, in file order. `pdftotext` is an optional runtime tool on the server: it runs without a shell on a 0600 temp file under `storage/app/tmp`, with a 60 s timeout and at most 1 MiB of text per file (the row's `truncated` is set beyond that). A scan or a broken PDF stores an empty text and adds nothing (there is no OCR); a missing `pdftotext` is logged once and stores nothing, so a later run retries. Detaching a file drops its text row (foreign-key cascade) and resyncs the material. `bm:materials-reindex` extracts the text of any PDF that has none before syncing each material, which backfills PDFs attached before file search existed.
|
||||
|
||||
Configuration, in the app's `config/search.yaml`: `search.driver`, `search.prefix`, `search.typesense.host`, `port`, `protocol` and `connection_timeout_seconds`. The API key goes only in the environment, `SUMMER_SEARCH__TYPESENSE__API_KEY`; with no key the engine is not configured, nothing is sent and every search answers 503.
|
||||
|
||||
## CLI commands
|
||||
@@ -389,7 +393,7 @@ Configuration, in the app's `config/search.yaml`: `search.driver`, `search.prefi
|
||||
./bin/bm bm:materials-reindex --fresh
|
||||
```
|
||||
|
||||
Ensures the `jz_bm_materials` collection and syncs every material through the same path as the sync after a save, ending with `Reindexed: N synced, M failed`; it exits non-zero when any material failed. `--fresh` first drops the collection and recreates it from `models.Material.SearchIndexSchema`: use it on the first deploy and after a schema change. Without an engine the command refuses with an error naming `search.driver` and `search.typesense.api_key`; `--if-configured` makes that a notice and a zero exit. The plugin schedules `bm:materials-reindex --if-configured` every night at 04:15 (`Schedule`, run by `serve`'s scheduler), because course renames, meeting renumbering, failed syncs and Typesense restarts can leave the index stale. Run it by hand after a Typesense restart.
|
||||
Ensures the `jz_bm_materials` collection and syncs every material through the same path as the sync after a save, first extracting the text of its PDFs that have none yet (a missing `pdftotext` is one warning, and the run goes on without file text), ending with `Reindexed: N synced, M failed`; it exits non-zero when any material failed. `--fresh` first drops the collection and recreates it from `models.Material.SearchIndexSchema`: use it on the first deploy and after a schema change. Without an engine the command refuses with an error naming `search.driver` and `search.typesense.api_key`; `--if-configured` makes that a notice and a zero exit. The plugin schedules `bm:materials-reindex --if-configured` every night at 04:15 (`Schedule`, run by `serve`'s scheduler), because course renames, meeting renumbering, failed syncs and Typesense restarts can leave the index stale. Run it by hand after a Typesense restart.
|
||||
|
||||
### bm:seed
|
||||
|
||||
|
||||
@@ -1,15 +1,20 @@
|
||||
package classes_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.golem15.com/golem15/summercms/modules/pact"
|
||||
"git.golem15.com/jakub/sm-bm-plugin/classes"
|
||||
"git.golem15.com/jakub/sm-bm-plugin/internal/searchtest"
|
||||
"git.golem15.com/jakub/sm-bm-plugin/jobs"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
func TestMaterialSearchGate(t *testing.T) {
|
||||
@@ -119,3 +124,60 @@ func materialSearchUnavailable(t *testing.T) {
|
||||
none.Sync(context.Background(), nil, 1)
|
||||
none.RemoveAfterCommit(context.Background(), nil, 1)
|
||||
}
|
||||
|
||||
func TestMaterialQueueExtract(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
// A nil indexer, or one without Enqueue, is a no-op.
|
||||
var none *classes.MaterialIndexer
|
||||
none.QueueExtract(ctx, nil, 3)
|
||||
(&classes.MaterialIndexer{}).QueueExtract(ctx, nil, 3)
|
||||
|
||||
var logs bytes.Buffer
|
||||
var got []pact.JobArgs
|
||||
var txs []*gorm.DB
|
||||
ix := &classes.MaterialIndexer{
|
||||
Log: slog.New(slog.NewTextHandler(&logs, nil)),
|
||||
Enqueue: func(_ context.Context, tx *gorm.DB, args pact.JobArgs) error {
|
||||
got = append(got, args)
|
||||
txs = append(txs, tx)
|
||||
return nil
|
||||
},
|
||||
}
|
||||
tx := &gorm.DB{}
|
||||
ix.QueueExtract(ctx, tx, 12)
|
||||
ix.QueueExtract(ctx, tx, 0)
|
||||
if len(got) != 1 {
|
||||
t.Fatalf("enqueued %d jobs, want 1 (id 0 queues nothing)", len(got))
|
||||
}
|
||||
args, ok := got[0].(jobs.ExtractMaterialTextArgs)
|
||||
if !ok || args.MaterialID != 12 || args.Kind() != "jz.bm.extract_material_text" {
|
||||
t.Errorf("args = %#v, want ExtractMaterialTextArgs{12} of kind jz.bm.extract_material_text", got[0])
|
||||
}
|
||||
if txs[0] != tx {
|
||||
t.Error("the job was not enqueued in the given transaction")
|
||||
}
|
||||
if logs.Len() != 0 {
|
||||
t.Errorf("a successful enqueue logged %q", logs.String())
|
||||
}
|
||||
|
||||
// A failing Enqueue is logged with the material id and not returned.
|
||||
ix.Enqueue = func(context.Context, *gorm.DB, pact.JobArgs) error { return errors.New("river is down") }
|
||||
ix.QueueExtract(ctx, tx, 12)
|
||||
if out := logs.String(); !strings.Contains(out, "could not queue the file text job") || !strings.Contains(out, "material_id=12") || !strings.Contains(out, "river is down") {
|
||||
t.Errorf("log = %q, want the failure with the material id", out)
|
||||
}
|
||||
|
||||
// The indexer of an app resolves conga at call time; without an app the
|
||||
// enqueue fails and is only logged.
|
||||
logs.Reset()
|
||||
app := classes.NewMaterialIndexer(nil)
|
||||
if app.Enqueue == nil {
|
||||
t.Fatal("NewMaterialIndexer left Enqueue nil")
|
||||
}
|
||||
app.Log = slog.New(slog.NewTextHandler(&logs, nil))
|
||||
app.QueueExtract(ctx, nil, 5)
|
||||
if !strings.Contains(logs.String(), "material_id=5") {
|
||||
t.Errorf("log = %q, want the failed enqueue logged", logs.String())
|
||||
}
|
||||
}
|
||||
|
||||
24
console/export_test.go
Normal file
24
console/export_test.go
Normal file
@@ -0,0 +1,24 @@
|
||||
package console
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"git.golem15.com/golem15/summercms/modules/bonfire"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// ReindexRun exposes one reindex walk to the console_test package.
|
||||
type ReindexRun struct{ r *reindexRun }
|
||||
|
||||
// NewReindexRun is a walk with deps' Backfill and the given sync.
|
||||
func NewReindexRun(deps MaterialsReindexDeps, sync func(id uint) error, out bonfire.Output) ReindexRun {
|
||||
return ReindexRun{r: &reindexRun{backfill: deps.Backfill, sync: sync, out: out}}
|
||||
}
|
||||
|
||||
// Material runs the per-material step.
|
||||
func (w ReindexRun) Material(ctx context.Context, db *gorm.DB, id uint) { w.r.material(ctx, db, id) }
|
||||
|
||||
// Counts are the run's counters and the tool-missing flag.
|
||||
func (w ReindexRun) Counts() (synced, failed int, noTool bool) {
|
||||
return w.r.synced, w.r.failed, w.r.noTool
|
||||
}
|
||||
@@ -28,6 +28,10 @@ var errSearchNotConfigured = errors.New("jz.bm: the search engine is not configu
|
||||
type MaterialsReindexDeps struct {
|
||||
DB func(ctx context.Context) (*gorm.DB, func(), error)
|
||||
Index *classes.MaterialIndexer
|
||||
// Backfill extracts the text of a material's PDFs that have none yet
|
||||
// (classes.EnsureFileTexts) before the material is synced. Nil skips
|
||||
// it.
|
||||
Backfill func(ctx context.Context, db *gorm.DB, materialID uint) error
|
||||
}
|
||||
|
||||
// MaterialsReindexCommand returns bm:materials-reindex: it ensures the
|
||||
@@ -41,7 +45,10 @@ func MaterialsReindexCommand(deps MaterialsReindexDeps) bonfire.Command {
|
||||
Description: "Sync every material into the search index",
|
||||
Long: `Upserts the search document of every material, drafts and private ones
|
||||
included (visibility is enforced in SQL, never by the index), through the same
|
||||
path as the sync after a save.
|
||||
path as the sync after a save. Before each material is synced, the text of
|
||||
its attached PDFs that have none yet is extracted with pdftotext, so PDFs
|
||||
attached before file search existed become searchable; a missing pdftotext
|
||||
(poppler-utils) is one warning and the run goes on without file text.
|
||||
|
||||
--fresh first drops the collection and recreates it from the material schema
|
||||
(models.Material.SearchIndexSchema). Use it on first deploy and after the
|
||||
@@ -103,35 +110,72 @@ func reindexMaterials(ctx context.Context, deps MaterialsReindexDeps, fresh, ifC
|
||||
}
|
||||
|
||||
marked := classes.MarkMaterialSearch(ctx)
|
||||
synced, failed := 0, 0
|
||||
run := &reindexRun{
|
||||
backfill: deps.Backfill,
|
||||
sync: func(id uint) error {
|
||||
return svc.Sync(marked, gdb, &models.Material{ID: id})
|
||||
},
|
||||
out: out,
|
||||
}
|
||||
var batch []models.Material
|
||||
err = gdb.WithContext(ctx).Model(&models.Material{}).
|
||||
Select("id").
|
||||
Order("id").
|
||||
FindInBatches(&batch, materialsReindexBatch, func(_ *gorm.DB, _ int) error {
|
||||
for i := range batch {
|
||||
id := batch[i].ID
|
||||
if err := svc.Sync(marked, gdb, &models.Material{ID: id}); err != nil {
|
||||
failed++
|
||||
out.Warning(fmt.Sprintf("material %d: %v", id, err))
|
||||
continue
|
||||
}
|
||||
synced++
|
||||
run.material(ctx, gdb, batch[i].ID)
|
||||
}
|
||||
return ctx.Err()
|
||||
}).Error
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
msg := fmt.Sprintf("Reindexed: %d synced, %d failed", synced, failed)
|
||||
if failed > 0 {
|
||||
msg := fmt.Sprintf("Reindexed: %d synced, %d failed", run.synced, run.failed)
|
||||
if run.failed > 0 {
|
||||
out.Error(msg)
|
||||
return fmt.Errorf("jz.bm: %d materials failed to sync", failed)
|
||||
return fmt.Errorf("jz.bm: %d materials failed to sync", run.failed)
|
||||
}
|
||||
out.Success(msg)
|
||||
return nil
|
||||
}
|
||||
|
||||
// pdfToolMissingWarning is printed once per run when pdftotext is missing.
|
||||
const pdfToolMissingWarning = "pdftotext not found: PDFs add no search text; install poppler-utils and rerun"
|
||||
|
||||
// reindexRun is the state of one reindex walk: the run's counters and
|
||||
// whether pdftotext turned out to be missing.
|
||||
type reindexRun struct {
|
||||
backfill func(ctx context.Context, db *gorm.DB, materialID uint) error
|
||||
sync func(id uint) error
|
||||
out bonfire.Output
|
||||
synced int
|
||||
failed int
|
||||
noTool bool
|
||||
}
|
||||
|
||||
// material backfills the file texts of material id, then syncs it. A
|
||||
// missing pdftotext warns once and stops the backfill for the rest of the
|
||||
// run; another backfill error warns for this material, which is still
|
||||
// synced.
|
||||
func (r *reindexRun) material(ctx context.Context, db *gorm.DB, id uint) {
|
||||
if r.backfill != nil && !r.noTool {
|
||||
err := r.backfill(ctx, db, id)
|
||||
switch {
|
||||
case errors.Is(err, classes.ErrPDFToolMissing):
|
||||
r.noTool = true
|
||||
r.out.Warning(pdfToolMissingWarning)
|
||||
case err != nil:
|
||||
r.out.Warning(fmt.Sprintf("material %d: file text: %v", id, err))
|
||||
}
|
||||
}
|
||||
if err := r.sync(id); err != nil {
|
||||
r.failed++
|
||||
r.out.Warning(fmt.Sprintf("material %d: %v", id, err))
|
||||
return
|
||||
}
|
||||
r.synced++
|
||||
}
|
||||
|
||||
// flagOn reports whether a bare flag was given.
|
||||
func flagOn(in bonfire.Input, name string) bool {
|
||||
v, _ := in.Flag(name)
|
||||
|
||||
@@ -3,8 +3,10 @@ package console_test
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
@@ -87,3 +89,83 @@ func TestMaterialsReindexFresh(t *testing.T) {
|
||||
t.Errorf("second run output %q, want Dropped", out.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestMaterialsReindexBackfill(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
var calls []string
|
||||
var synced []uint
|
||||
sync := func(id uint) error {
|
||||
calls = append(calls, "sync "+strconv.FormatUint(uint64(id), 10))
|
||||
synced = append(synced, id)
|
||||
return nil
|
||||
}
|
||||
|
||||
// pdftotext missing on the first material: one warning, no more
|
||||
// backfill calls, and every material is still synced.
|
||||
deps := console.MaterialsReindexDeps{Backfill: func(_ context.Context, _ *gorm.DB, id uint) error {
|
||||
calls = append(calls, "backfill "+strconv.FormatUint(uint64(id), 10))
|
||||
return classes.ErrPDFToolMissing
|
||||
}}
|
||||
var stdout, stderr bytes.Buffer
|
||||
run := console.NewReindexRun(deps, sync, bonfire.NewOutput(strings.NewReader(""), &stdout, &stderr))
|
||||
for _, id := range []uint{1, 2, 3} {
|
||||
run.Material(ctx, nil, id)
|
||||
}
|
||||
if want := []string{"backfill 1", "sync 1", "sync 2", "sync 3"}; !slices.Equal(calls, want) {
|
||||
t.Errorf("calls = %v, want %v", calls, want)
|
||||
}
|
||||
if !slices.Equal(synced, []uint{1, 2, 3}) {
|
||||
t.Errorf("synced = %v, want all three", synced)
|
||||
}
|
||||
all := stdout.String() + stderr.String()
|
||||
if n := strings.Count(all, "pdftotext not found"); n != 1 {
|
||||
t.Errorf("the tool warning was printed %d times, want once:\n%s", n, all)
|
||||
}
|
||||
if !strings.Contains(all, "install poppler-utils") {
|
||||
t.Errorf("output %q, want the poppler-utils hint", all)
|
||||
}
|
||||
if s, f, noTool := run.Counts(); s != 3 || f != 0 || !noTool {
|
||||
t.Errorf("counts = %d synced %d failed noTool %v, want 3 0 true", s, f, noTool)
|
||||
}
|
||||
|
||||
// Another backfill error warns for that material only; it is still
|
||||
// synced and the next material is backfilled. A failed sync counts.
|
||||
calls, synced = nil, nil
|
||||
stdout.Reset()
|
||||
stderr.Reset()
|
||||
deps.Backfill = func(_ context.Context, _ *gorm.DB, id uint) error {
|
||||
calls = append(calls, "backfill "+strconv.FormatUint(uint64(id), 10))
|
||||
if id == 1 {
|
||||
return errors.New("blob gone")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
failing := func(id uint) error {
|
||||
if err := sync(id); err != nil || id == 3 {
|
||||
return errors.New("engine down")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
run = console.NewReindexRun(deps, failing, bonfire.NewOutput(strings.NewReader(""), &stdout, &stderr))
|
||||
for _, id := range []uint{1, 2, 3} {
|
||||
run.Material(ctx, nil, id)
|
||||
}
|
||||
if want := []string{"backfill 1", "sync 1", "backfill 2", "sync 2", "backfill 3", "sync 3"}; !slices.Equal(calls, want) {
|
||||
t.Errorf("calls = %v, want %v", calls, want)
|
||||
}
|
||||
all = stdout.String() + stderr.String()
|
||||
if !strings.Contains(all, "material 1: file text: blob gone") || strings.Contains(all, "pdftotext not found") {
|
||||
t.Errorf("output %q, want the material 1 file text warning only", all)
|
||||
}
|
||||
if s, f, noTool := run.Counts(); s != 2 || f != 1 || noTool {
|
||||
t.Errorf("counts = %d synced %d failed noTool %v, want 2 1 false", s, f, noTool)
|
||||
}
|
||||
|
||||
// No Backfill: only syncs.
|
||||
calls = nil
|
||||
run = console.NewReindexRun(console.MaterialsReindexDeps{}, sync, bonfire.NewOutput(strings.NewReader(""), &stdout, &stderr))
|
||||
run.Material(ctx, nil, 9)
|
||||
if !slices.Equal(calls, []string{"sync 9"}) {
|
||||
t.Errorf("calls without Backfill = %v", calls)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -208,7 +208,8 @@ func sizeMB(n int64) string {
|
||||
}
|
||||
|
||||
// DetachMaterialFile answers DELETE /materials/{id}/files/{file}: the file
|
||||
// row goes, then its blob and thumbnails. A file of another material is a
|
||||
// row goes (its extracted text with it), the material is resynced, then the
|
||||
// blob and thumbnails go. A file of another material is a
|
||||
// 404.
|
||||
func (h *Handlers) DetachMaterialFile(w http.ResponseWriter, r *http.Request) {
|
||||
db, ok := h.db(w, r)
|
||||
@@ -242,6 +243,9 @@ func (h *Handlers) DetachMaterialFile(w http.ResponseWriter, r *http.Request) {
|
||||
writeOpaque500(w)
|
||||
return
|
||||
}
|
||||
// The file's text row went with it (foreign-key cascade); the resync
|
||||
// drops the text from the material's document.
|
||||
h.MaterialIndex.Sync(r.Context(), db, m.ID)
|
||||
// The row is gone, so the file is no longer served; a blob left behind
|
||||
// by a failed delete is only disk space.
|
||||
_ = attach.DeleteKeys(r.Context(), bucket, attach.BlobKeys(f))
|
||||
|
||||
@@ -134,6 +134,10 @@ func (h *Handlers) AttachMaterialInline(w http.ResponseWriter, r *http.Request)
|
||||
err := lagoon.Transaction(r.Context(), db, func(ctx context.Context, tx *gorm.DB) error {
|
||||
f, why, err := storeMaterialFile(ctx, tx, bucket, m, name, bytes.NewReader(data), lim)
|
||||
stored, refusal = f, why
|
||||
if err == nil && why == "" && f != nil {
|
||||
// The text job joins the transaction and syncs the material.
|
||||
h.MaterialIndex.QueueExtract(ctx, tx, m.ID)
|
||||
}
|
||||
return err
|
||||
})
|
||||
if err != nil || refusal != "" {
|
||||
@@ -302,6 +306,8 @@ func (h *Handlers) SignedMaterialUpload(w http.ResponseWriter, r *http.Request)
|
||||
refusal = singleFilePart
|
||||
return errUploadRefused
|
||||
}
|
||||
// The text job joins the transaction and syncs the material.
|
||||
h.MaterialIndex.QueueExtract(ctx, tx, m.ID)
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
|
||||
@@ -18,10 +18,12 @@ var (
|
||||
)
|
||||
|
||||
// MaterialIndex keeps the materials search index current once a write
|
||||
// commits (classes.MaterialIndexer). Its calls never fail the write.
|
||||
// commits (classes.MaterialIndexer). QueueExtract enqueues the file text job
|
||||
// inside tx. Its calls never fail the write.
|
||||
type MaterialIndex interface {
|
||||
SyncAfterCommit(ctx context.Context, tx *gorm.DB, id uint)
|
||||
RemoveAfterCommit(ctx context.Context, tx *gorm.DB, id uint)
|
||||
QueueExtract(ctx context.Context, tx *gorm.DB, id uint)
|
||||
}
|
||||
|
||||
type materialIndexHolder struct{ ix MaterialIndex }
|
||||
@@ -40,12 +42,17 @@ func installedMaterialIndex() MaterialIndex {
|
||||
return h.ix
|
||||
}
|
||||
|
||||
// FormAfterCreate syncs the new material once the save transaction commits.
|
||||
// FormAfterCreate queues the file text job in the save transaction (the
|
||||
// form's deferred files commit in the same one) and syncs the new material
|
||||
// once it commits.
|
||||
func (materialsAdmin) FormAfterCreate(ctx context.Context, model any) error {
|
||||
return materialAfterCommit(ctx, model, false)
|
||||
}
|
||||
|
||||
// FormAfterUpdate syncs the material once the save transaction commits.
|
||||
// FormAfterUpdate queues the file text job in the save transaction and
|
||||
// syncs the material once it commits. A save that added no file queues a
|
||||
// job that finds no new PDF and only resyncs; a file removed in the form
|
||||
// leaves the document through the cascade and this sync.
|
||||
func (materialsAdmin) FormAfterUpdate(ctx context.Context, model any) error {
|
||||
return materialAfterCommit(ctx, model, false)
|
||||
}
|
||||
@@ -70,8 +77,9 @@ func materialAfterCommit(ctx context.Context, model any, remove bool) error {
|
||||
tx, _ := cabana.TxFromContext(ctx)
|
||||
if remove {
|
||||
ix.RemoveAfterCommit(ctx, tx, m.ID)
|
||||
} else {
|
||||
ix.SyncAfterCommit(ctx, tx, m.ID)
|
||||
return nil
|
||||
}
|
||||
ix.QueueExtract(ctx, tx, m.ID)
|
||||
ix.SyncAfterCommit(ctx, tx, m.ID)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -26,6 +26,11 @@ func (r *recordingIndex) RemoveAfterCommit(_ context.Context, tx *gorm.DB, id ui
|
||||
r.txs = append(r.txs, tx)
|
||||
}
|
||||
|
||||
func (r *recordingIndex) QueueExtract(_ context.Context, tx *gorm.DB, id uint) {
|
||||
r.calls = append(r.calls, "queue "+itoa(id))
|
||||
r.txs = append(r.txs, tx)
|
||||
}
|
||||
|
||||
func itoa(id uint) string { return strconv.FormatUint(uint64(id), 10) }
|
||||
|
||||
func TestMaterialsAdminHooks(t *testing.T) {
|
||||
@@ -53,7 +58,9 @@ func TestMaterialsAdminHooks(t *testing.T) {
|
||||
if err := c.FormAfterDelete(ctx, m); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if want := []string{"sync 7", "sync 7", "remove 7"}; !slices.Equal(rec.calls, want) {
|
||||
// Create and update queue the file text job (in the save transaction)
|
||||
// before the after-commit sync; delete queues nothing.
|
||||
if want := []string{"queue 7", "sync 7", "queue 7", "sync 7", "remove 7"}; !slices.Equal(rec.calls, want) {
|
||||
t.Errorf("calls = %v, want %v", rec.calls, want)
|
||||
}
|
||||
// Outside a cabana write there is no transaction: the index then runs
|
||||
|
||||
178
material_file_text_postgres_test.go
Normal file
178
material_file_text_postgres_test.go
Normal file
@@ -0,0 +1,178 @@
|
||||
package bm_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os/exec"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
userclasses "git.golem15.com/golem15/sm-user-plugin/classes"
|
||||
"git.golem15.com/golem15/summercms/modules/surf"
|
||||
"git.golem15.com/jakub/sm-bm-plugin/classes"
|
||||
"git.golem15.com/jakub/sm-bm-plugin/internal/pdftest"
|
||||
"git.golem15.com/jakub/sm-bm-plugin/internal/searchtest"
|
||||
"git.golem15.com/jakub/sm-bm-plugin/jobs"
|
||||
"gocloud.dev/blob"
|
||||
_ "gocloud.dev/blob/memblob"
|
||||
)
|
||||
|
||||
// TestMaterialFileTextAgainstDatabase drives file search end to end against
|
||||
// a migrated database and a Typesense stand-in: an inline PDF attach queues
|
||||
// one extraction job in its transaction, the extraction stores the PDF's
|
||||
// text, the sync sends it as file_texts, and a detach drops the text row
|
||||
// (cascade) and resyncs the document without it.
|
||||
func TestMaterialFileTextAgainstDatabase(t *testing.T) {
|
||||
stub := &searchtest.Stub{}
|
||||
app, plugins, gdb := migratedWith(t, searchtest.Start(t, stub)...)
|
||||
if _, err := exec.LookPath("pdftotext"); err != nil {
|
||||
t.Skip("pdftotext is not installed (poppler-utils)")
|
||||
}
|
||||
bucket, err := blob.OpenBucket(t.Context(), "mem://")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { bucket.Close() })
|
||||
if err := app.Publish(bucket); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
handler, err := surf.Assemble(app, plugins)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
managerID := createUser(t, gdb, "files-manager@example.test")
|
||||
var groupID uint
|
||||
if err := gdb.Raw(`
|
||||
INSERT INTO user_groups (name, code, permissions, created_at, updated_at)
|
||||
VALUES ('BM materials', 'bm-materials', '{"jz.bm.access_materials":1}', NOW(), NOW())
|
||||
RETURNING id`).Scan(&groupID).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := gdb.Exec(`INSERT INTO users_groups (user_id, user_group_id) VALUES (?, ?)`, managerID, groupID).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
secret, token, err := userclasses.MintAPIToken("usr_", managerID, "BM file text", []string{"read", "write"}, nil, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := gdb.Create(token).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
send := func(method, path string, payload any) (int, map[string]any) {
|
||||
t.Helper()
|
||||
var buf bytes.Buffer
|
||||
if payload != nil {
|
||||
if err := json.NewEncoder(&buf).Encode(payload); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
req := httptest.NewRequest(method, path, &buf)
|
||||
req.Header.Set("Authorization", "Bearer "+secret)
|
||||
if payload != nil {
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
}
|
||||
rec := httptest.NewRecorder()
|
||||
handler.ServeHTTP(rec, req)
|
||||
var out map[string]any
|
||||
if rec.Body.Len() > 0 {
|
||||
_ = json.Unmarshal(rec.Body.Bytes(), &out)
|
||||
}
|
||||
return rec.Code, out
|
||||
}
|
||||
|
||||
code, out := send(http.MethodPost, "/api/v1/bm/materials", map[string]any{"title": "Wykład z PDF", "type": "notatka"})
|
||||
if code != http.StatusCreated {
|
||||
t.Fatalf("POST material = %d %v", code, out)
|
||||
}
|
||||
materialID := id(obj(out["data"])["id"])
|
||||
n, _ := strconv.ParseUint(materialID, 10, 64)
|
||||
mid := uint(n)
|
||||
|
||||
const phrase = "Globalizacja i demokracja"
|
||||
code, out = send(http.MethodPost, "/api/v1/bm/materials/"+materialID+"/files/inline", map[string]any{
|
||||
"name": "wyklad.pdf",
|
||||
"content_base64": base64.StdEncoding.EncodeToString(pdftest.OnePage(phrase)),
|
||||
})
|
||||
if code != http.StatusCreated {
|
||||
t.Fatalf("inline attach = %d %v", code, out)
|
||||
}
|
||||
var queued int64
|
||||
if err := gdb.Raw(`SELECT COUNT(*) FROM river_job WHERE kind = ?`, jobs.ExtractMaterialTextKind).Scan(&queued).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if queued != 1 {
|
||||
t.Errorf("river_job rows of kind %s = %d, want 1", jobs.ExtractMaterialTextKind, queued)
|
||||
}
|
||||
|
||||
pdf := classes.DefaultPDFText()
|
||||
pdf.TempDir = t.TempDir()
|
||||
added, err := classes.EnsureFileTexts(t.Context(), gdb, bucket, pdf, mid)
|
||||
if err != nil || added != 1 {
|
||||
t.Fatalf("EnsureFileTexts = %d, %v, want 1 row", added, err)
|
||||
}
|
||||
var texts []string
|
||||
if err := gdb.Raw(`SELECT text FROM jz_bm_material_file_texts WHERE material_id = ?`, mid).Scan(&texts).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(texts) != 1 || !strings.Contains(texts[0], phrase) {
|
||||
t.Fatalf("text rows = %q, want one holding %q", texts, phrase)
|
||||
}
|
||||
// A second run finds nothing new.
|
||||
if again, err := classes.EnsureFileTexts(t.Context(), gdb, bucket, pdf, mid); err != nil || again != 0 {
|
||||
t.Errorf("second EnsureFileTexts = %d, %v, want 0", again, err)
|
||||
}
|
||||
|
||||
lastDoc := func() map[string]any {
|
||||
t.Helper()
|
||||
var doc map[string]any
|
||||
for _, d := range stub.Imported() {
|
||||
if d["id"] == materialID {
|
||||
doc = d
|
||||
}
|
||||
}
|
||||
if doc == nil {
|
||||
t.Fatalf("material %s was never upserted", materialID)
|
||||
}
|
||||
return doc
|
||||
}
|
||||
ix := classes.NewMaterialIndexer(app)
|
||||
ix.Sync(t.Context(), gdb, mid)
|
||||
fileTexts, _ := lastDoc()["file_texts"].([]any)
|
||||
if len(fileTexts) != 1 || !strings.Contains(fileTexts[0].(string), phrase) {
|
||||
t.Errorf("upserted file_texts = %v, want the PDF's text", lastDoc()["file_texts"])
|
||||
}
|
||||
|
||||
var fileIDs []uint
|
||||
if err := gdb.Raw(`SELECT id FROM system_files WHERE attachment_id = ? AND field = 'files'`, materialID).Scan(&fileIDs).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(fileIDs) != 1 {
|
||||
t.Fatalf("files = %v, want one", fileIDs)
|
||||
}
|
||||
before := len(stub.Imported())
|
||||
code, out = send(http.MethodDelete, "/api/v1/bm/materials/"+materialID+"/files/"+strconv.FormatUint(uint64(fileIDs[0]), 10), nil)
|
||||
if code != http.StatusNoContent {
|
||||
t.Fatalf("detach = %d %v", code, out)
|
||||
}
|
||||
var left int64
|
||||
if err := gdb.Raw(`SELECT COUNT(*) FROM jz_bm_material_file_texts WHERE material_id = ?`, mid).Scan(&left).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if left != 0 {
|
||||
t.Errorf("text rows after detach = %d, want 0 (cascade)", left)
|
||||
}
|
||||
if len(stub.Imported()) <= before {
|
||||
t.Fatal("the detach did not resync the material")
|
||||
}
|
||||
if _, ok := lastDoc()["file_texts"]; ok {
|
||||
t.Errorf("the document after detach still has file_texts: %v", lastDoc())
|
||||
}
|
||||
if slices.Contains(stub.Removed(), materialID) {
|
||||
t.Error("the detach removed the material's document instead of resyncing it")
|
||||
}
|
||||
}
|
||||
41
plugin.go
41
plugin.go
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"embed"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io/fs"
|
||||
"time"
|
||||
|
||||
@@ -11,6 +12,7 @@ import (
|
||||
"git.golem15.com/golem15/summercms/modules/bonfire"
|
||||
"git.golem15.com/golem15/summercms/modules/conga"
|
||||
"git.golem15.com/golem15/summercms/modules/lagoon"
|
||||
"git.golem15.com/golem15/summercms/modules/lagoon/attach"
|
||||
"git.golem15.com/golem15/summercms/modules/lighthouse"
|
||||
"git.golem15.com/golem15/summercms/modules/lighthouse/centrifugo"
|
||||
"git.golem15.com/golem15/summercms/modules/pact"
|
||||
@@ -19,6 +21,7 @@ import (
|
||||
"git.golem15.com/jakub/sm-bm-plugin/console"
|
||||
"git.golem15.com/jakub/sm-bm-plugin/controllers"
|
||||
"github.com/go-gormigrate/gormigrate/v2"
|
||||
"gocloud.dev/blob"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
@@ -131,7 +134,7 @@ func (p *Plugin) Commands() []bonfire.Command {
|
||||
cmds = append(cmds, console.ChatSyncCommand(p.withDB, func() (string, string) {
|
||||
return p.chatGroup(), p.chatModeratorGroup()
|
||||
}))
|
||||
cmds = append(cmds, console.MaterialsReindexCommand(console.MaterialsReindexDeps{DB: p.publishedDB, Index: p.materialIndex}))
|
||||
cmds = append(cmds, console.MaterialsReindexCommand(console.MaterialsReindexDeps{DB: p.publishedDB, Index: p.materialIndex, Backfill: p.backfillFileTexts}))
|
||||
return append(cmds, centrifugo.Commands(p.app)...)
|
||||
}
|
||||
|
||||
@@ -222,6 +225,42 @@ func (p *Plugin) publishedDB(ctx context.Context) (*gorm.DB, func(), error) {
|
||||
return gdb, func() { _ = sqlDB.Close() }, nil
|
||||
}
|
||||
|
||||
// backfillFileTexts extracts the text of material id's PDFs that have none
|
||||
// yet, for bm:materials-reindex (D-04: PDFs attached before file search).
|
||||
func (p *Plugin) backfillFileTexts(ctx context.Context, db *gorm.DB, id uint) error {
|
||||
bucket, err := p.uploadsBucket(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = classes.EnsureFileTexts(ctx, db, bucket, classes.DefaultPDFText(), id)
|
||||
return err
|
||||
}
|
||||
|
||||
// uploadsBucket returns the uploads bucket published on the app: the one
|
||||
// serve published or, when a console command runs, one opened from
|
||||
// storage.uploads.bucket_url and published. It is an error when no bucket
|
||||
// is configured.
|
||||
func (p *Plugin) uploadsBucket(ctx context.Context) (*blob.Bucket, error) {
|
||||
if p.app == nil {
|
||||
return nil, errors.New("jz.bm: app is nil")
|
||||
}
|
||||
if b, ok := p.app.Lookup[*blob.Bucket](); ok && b != nil {
|
||||
return b, nil
|
||||
}
|
||||
b, err := attach.OpenBucket(ctx, p.app.Config)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("jz.bm: the uploads bucket is not configured: %w", err)
|
||||
}
|
||||
if err := attach.Publish(p.app, b); err != nil {
|
||||
_ = b.Close()
|
||||
if pub, ok := p.app.Lookup[*blob.Bucket](); ok && pub != nil {
|
||||
return pub, nil
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
return b, nil
|
||||
}
|
||||
|
||||
func init() {
|
||||
party.Register(&Plugin{})
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user