diff --git a/db/go.mod b/db/go.mod index df444bd7..8efc8e6e 100644 --- a/db/go.mod +++ b/db/go.mod @@ -32,7 +32,7 @@ require ( github.com/ganigeorgiev/fexpr v0.5.0 // indirect github.com/go-ap/activitypub v0.0.0-20250905102448-e9df599e4528 github.com/go-fed/httpsig v1.1.0 - github.com/go-ozzo/ozzo-validation/v4 v4.3.0 // indirect + github.com/go-ozzo/ozzo-validation/v4 v4.3.0 github.com/google/uuid v1.6.0 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/mattn/go-colorable v0.1.14 // indirect diff --git a/db/hooks/trails.go b/db/hooks/trails.go index a3045b46..40bcf4ce 100644 --- a/db/hooks/trails.go +++ b/db/hooks/trails.go @@ -20,6 +20,9 @@ func CreateTrailHandler(client meilisearch.ServiceManager) func(e *core.RecordEv if err != nil { return err } + if err := util.SavePolyline(e.App, record); err != nil { + log.Printf("failed to save polyline for trail %s: %v", record.Id, err) + } if err := util.IndexTrails(e.App, []*core.Record{record}, client); err != nil { return err } @@ -62,6 +65,13 @@ func UpdateTrailHandler(client meilisearch.ServiceManager) func(e *core.RecordEv if err != nil { return err } + + if record.GetString("gpx") != record.Original().GetString("gpx") { + if err := util.SavePolyline(e.App, record); err != nil { + log.Printf("failed to save polyline for trail %s: %v", record.Id, err) + } + } + err = util.UpdateTrail(e.App, record, userActor, client) if err != nil { return err diff --git a/db/main.go b/db/main.go index aa1a1800..046455e7 100644 --- a/db/main.go +++ b/db/main.go @@ -7,6 +7,7 @@ import ( "strings" "github.com/meilisearch/meilisearch-go" + "github.com/pocketbase/dbx" "github.com/pocketbase/pocketbase" "github.com/pocketbase/pocketbase/core" "github.com/pocketbase/pocketbase/plugins/migratecmd" @@ -213,10 +214,55 @@ func registerCronJobs(app core.App, client meilisearch.ServiceManager) { func initData(app core.App, client meilisearch.ServiceManager) error { initCategories(app) initMeilisearchConfig(client) - go initMeilisearchDocuments(app, client) + go func() { + backfillPolylines(app) + initMeilisearchDocuments(app, client) + }() return nil } +func backfillPolylines(app core.App) { + const pageSize int64 = 100 + var lastID string + var processed int + var failed int + + log.Printf("backfill polyline started") + defer func() { + log.Printf("backfill polyline completed: processed=%d failed=%d", processed, failed) + }() + + for { + trails := []*core.Record{} + query := app.RecordQuery("trails"). + AndWhere(dbx.NewExp("(polyline IS NULL OR polyline = '') AND gpx != ''")). + OrderBy("id ASC"). + Limit(pageSize) + + if lastID != "" { + query = query.AndWhere(dbx.NewExp("id > {:lastID}", dbx.Params{"lastID": lastID})) + } + + err := query.All(&trails) + if err != nil { + log.Printf("backfill polyline query failed after trail %q: %v", lastID, err) + break + } + if len(trails) == 0 { + break + } + for _, r := range trails { + if err := util.SavePolyline(app, r); err != nil { + failed++ + log.Printf("backfill polyline failed for trail %s (%q), gpx=%q: %v", r.Id, r.GetString("name"), r.GetString("gpx"), err) + } else { + processed++ + } + lastID = r.Id + } + } +} + func initCategories(app core.App) error { query := app.RecordQuery("categories") records := []*core.Record{} diff --git a/db/migrations/1748000000_updated_trails_polyline.go b/db/migrations/1748000000_updated_trails_polyline.go new file mode 100644 index 00000000..d82d6e9c --- /dev/null +++ b/db/migrations/1748000000_updated_trails_polyline.go @@ -0,0 +1,35 @@ +package migrations + +import ( + "github.com/pocketbase/pocketbase/core" + m "github.com/pocketbase/pocketbase/migrations" +) + +func init() { + m.Register(func(app core.App) error { + collection, err := app.FindCollectionByNameOrId("e864strfxo14pm4") + if err != nil { + return err + } + + if collection.Fields.GetByName("polyline") != nil { + return nil + } + + field := &core.TextField{} + field.Name = "polyline" + field.Required = false + collection.Fields.Add(field) + + return app.Save(collection) + }, func(app core.App) error { + collection, err := app.FindCollectionByNameOrId("e864strfxo14pm4") + if err != nil { + return err + } + + collection.Fields.RemoveByName("polyline") + + return app.Save(collection) + }) +} diff --git a/db/routes/remote_list.go b/db/routes/remote_list.go index 464bdb60..2e08aba7 100644 --- a/db/routes/remote_list.go +++ b/db/routes/remote_list.go @@ -51,8 +51,17 @@ func RemoteListGet(e *core.RequestEvent) error { } } else { updatedAt := record.GetDateTime("updated").Time() - if time.Now().UTC().Sub(updatedAt) > 60*time.Minute { - go performFullListSync(e.App, ctx, e.Request.URL, record) + + iri := record.GetString("iri") + if time.Now().UTC().Sub(updatedAt) > remoteSyncThreshold { + if _, alreadySyncing := listSyncing.LoadOrStore(iri, struct{}{}); !alreadySyncing { + urlCopy := *e.Request.URL + bgCtx := context.WithValue(context.Background(), "actor", ctx.Value("actor")) + go func() { + defer listSyncing.Delete(iri) + performFullListSync(e.App, bgCtx, &urlCopy, record) + }() + } } } } else { diff --git a/db/routes/remote_trail.go b/db/routes/remote_trail.go index c98dc496..01bd8f90 100644 --- a/db/routes/remote_trail.go +++ b/db/routes/remote_trail.go @@ -8,10 +8,13 @@ import ( "io" "net/http" "net/url" + "os" "path" "pocketbase/federation" "pocketbase/util" + "strconv" "strings" + "sync" "time" "github.com/pocketbase/dbx" @@ -19,6 +22,21 @@ import ( "github.com/pocketbase/pocketbase/tools/filesystem" ) +// remoteSyncThreshold is the minimum age of a remote record before a background sync is triggered. +// Configurable via POCKETBASE_FEDERATION_SYNC_INTERVAL (minutes). Default: 60. +var remoteSyncThreshold = func() time.Duration { + if v := os.Getenv("POCKETBASE_FEDERATION_SYNC_INTERVAL"); v != "" { + if minutes, err := strconv.Atoi(v); err == nil && minutes > 0 { + return time.Duration(minutes) * time.Minute + } + } + return 60 * time.Minute +}() + +// trailSyncing and listSyncing track IRIs currently being synced to prevent concurrent duplicate syncs. +var trailSyncing sync.Map +var listSyncing sync.Map + // --- Main Handler --- func RemoteTrailGet(e *core.RequestEvent) error { @@ -64,8 +82,17 @@ func RemoteTrailGet(e *core.RequestEvent) error { } else { // We already have it locally. Show and update background. updatedAt := record.GetDateTime("updated").Time() - if time.Now().UTC().Sub(updatedAt) > 60*time.Minute { - go performFullSync(e.App, ctx, e.Request.URL, record) + + iri := record.GetString("iri") + if time.Now().UTC().Sub(updatedAt) > remoteSyncThreshold { + if _, alreadySyncing := trailSyncing.LoadOrStore(iri, struct{}{}); !alreadySyncing { + urlCopy := *e.Request.URL + bgCtx := context.WithValue(context.Background(), "actor", ctx.Value("actor")) + go func() { + defer trailSyncing.Delete(iri) + performFullSync(e.App, bgCtx, &urlCopy, record) + }() + } } } } else { diff --git a/db/util/meilisearch.go b/db/util/meilisearch.go index f57a0750..92e2a86e 100644 --- a/db/util/meilisearch.go +++ b/db/util/meilisearch.go @@ -13,8 +13,6 @@ import ( "github.com/meilisearch/meilisearch-go" "github.com/pocketbase/pocketbase/core" - "github.com/tkrajina/gpxgo/gpx" - "github.com/twpayne/go-polyline" ) func documentFromTrailRecord(app core.App, r *core.Record, author *core.Record, includeShares bool) (map[string]interface{}, error) { @@ -41,11 +39,6 @@ func documentFromTrailRecord(app core.App, r *core.Record, author *core.Record, category = trailCategory.GetString("name") } - polyline, err := getPolyline(app, r) - if err != nil { - polyline = "" - } - domain := "" if !author.GetBool("isLocal") { domain = author.GetString("domain") @@ -72,7 +65,7 @@ func documentFromTrailRecord(app core.App, r *core.Record, author *core.Record, "thumbnail": thumbnail, "gpx": r.GetString("gpx"), "tags": tags, - "polyline": polyline, + "polyline": r.GetString("polyline"), "domain": domain, "iri": r.GetString("iri"), "_geo": map[string]float64{ @@ -128,46 +121,6 @@ func difficultyToNumber(difficulty string) int32 { return 0 } -func getPolyline(app core.App, r *core.Record) (string, error) { - gpxPath := r.GetString("gpx") - if len(gpxPath) == 0 { - return "", nil - } - avatarKey := r.BaseFilesPath() + "/" + gpxPath - fsys, err := app.NewFilesystem() - if err != nil { - return "", err - } - defer fsys.Close() - - gpxFile, err := fsys.GetReader(avatarKey) - if err != nil { - return "", err - } - defer gpxFile.Close() - - content := new(bytes.Buffer) - _, err = io.Copy(content, gpxFile) - if err != nil { - return "", err - } - gpxData, err := gpx.Parse(content) - if err != nil { - return "", err - } - - gpxData.SimplifyTracks(50) - coordinates := make([][]float64, 4) - for _, trk := range gpxData.Tracks { - for _, seg := range trk.Segments { - for _, pt := range seg.Points { - coordinates = append(coordinates, []float64{pt.Latitude, pt.Longitude}) - } - } - } - return string(polyline.EncodeCoords(coordinates)), nil -} - func documentFromListRecord(r *core.Record, author *core.Record, includeShares bool) (map[string]any, error) { totalElevationGain := 0.0 @@ -359,18 +312,10 @@ func UpdateTrail(app core.App, r *core.Record, author *core.Record, client meili } documents := []map[string]interface{}{doc} - task, err := client.Index("trails").UpdateDocuments(documents, nil) - - if err != nil { + if _, err = client.Index("trails").UpdateDocuments(documents, nil); err != nil { return err } - interval := 500 * time.Millisecond - _, err = client.WaitForTask(task.TaskUID, interval) - if err != nil { - return fmt.Errorf("meilisearch update trail: error waiting for task completion: %v", err) - } - return nil } diff --git a/db/util/polyline.go b/db/util/polyline.go new file mode 100644 index 00000000..4b782909 --- /dev/null +++ b/db/util/polyline.go @@ -0,0 +1,64 @@ +package util + +import ( + "bytes" + "fmt" + "io" + + "github.com/pocketbase/pocketbase/core" + "github.com/tkrajina/gpxgo/gpx" + "github.com/twpayne/go-polyline" +) + +func ComputePolyline(app core.App, r *core.Record) (string, error) { + gpxPath := r.GetString("gpx") + if len(gpxPath) == 0 { + return "", nil + } + + fsys, err := app.NewFilesystem() + if err != nil { + return "", fmt.Errorf("open filesystem: %w", err) + } + defer fsys.Close() + + gpxFilePath := r.BaseFilesPath() + "/" + gpxPath + gpxFile, err := fsys.GetReader(gpxFilePath) + if err != nil { + return "", fmt.Errorf("open gpx file %q: %w", gpxFilePath, err) + } + defer gpxFile.Close() + + content := new(bytes.Buffer) + if _, err = io.Copy(content, gpxFile); err != nil { + return "", fmt.Errorf("read gpx file %q: %w", gpxFilePath, err) + } + + gpxData, err := gpx.Parse(content) + if err != nil { + return "", fmt.Errorf("parse gpx file %q: %w", gpxFilePath, err) + } + + gpxData.SimplifyTracks(50) + coordinates := make([][]float64, 0) + for _, trk := range gpxData.Tracks { + for _, seg := range trk.Segments { + for _, pt := range seg.Points { + coordinates = append(coordinates, []float64{pt.Latitude, pt.Longitude}) + } + } + } + return string(polyline.EncodeCoords(coordinates)), nil +} + +func SavePolyline(app core.App, r *core.Record) error { + encoded, err := ComputePolyline(app, r) + if err != nil { + return err + } + r.Set("polyline", encoded) + if err := app.UnsafeWithoutHooks().Save(r); err != nil { + return fmt.Errorf("save trail polyline: %w", err) + } + return nil +}