fix: reduce Meilisearch load, debounce federation sync (#1012)
* optimize meili trail index * several fixes --------- Co-authored-by: Flomp <Flomp@users.noreply.github.com>
This commit is contained in:
@@ -32,7 +32,7 @@ require (
|
|||||||
github.com/ganigeorgiev/fexpr v0.5.0 // indirect
|
github.com/ganigeorgiev/fexpr v0.5.0 // indirect
|
||||||
github.com/go-ap/activitypub v0.0.0-20250905102448-e9df599e4528
|
github.com/go-ap/activitypub v0.0.0-20250905102448-e9df599e4528
|
||||||
github.com/go-fed/httpsig v1.1.0
|
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/google/uuid v1.6.0 // indirect
|
||||||
github.com/inconshreveable/mousetrap v1.1.0 // indirect
|
github.com/inconshreveable/mousetrap v1.1.0 // indirect
|
||||||
github.com/mattn/go-colorable v0.1.14 // indirect
|
github.com/mattn/go-colorable v0.1.14 // indirect
|
||||||
|
|||||||
@@ -20,6 +20,9 @@ func CreateTrailHandler(client meilisearch.ServiceManager) func(e *core.RecordEv
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
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 {
|
if err := util.IndexTrails(e.App, []*core.Record{record}, client); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -62,6 +65,13 @@ func UpdateTrailHandler(client meilisearch.ServiceManager) func(e *core.RecordEv
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
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)
|
err = util.UpdateTrail(e.App, record, userActor, client)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
|
|||||||
48
db/main.go
48
db/main.go
@@ -7,6 +7,7 @@ import (
|
|||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
"github.com/meilisearch/meilisearch-go"
|
"github.com/meilisearch/meilisearch-go"
|
||||||
|
"github.com/pocketbase/dbx"
|
||||||
"github.com/pocketbase/pocketbase"
|
"github.com/pocketbase/pocketbase"
|
||||||
"github.com/pocketbase/pocketbase/core"
|
"github.com/pocketbase/pocketbase/core"
|
||||||
"github.com/pocketbase/pocketbase/plugins/migratecmd"
|
"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 {
|
func initData(app core.App, client meilisearch.ServiceManager) error {
|
||||||
initCategories(app)
|
initCategories(app)
|
||||||
initMeilisearchConfig(client)
|
initMeilisearchConfig(client)
|
||||||
go initMeilisearchDocuments(app, client)
|
go func() {
|
||||||
|
backfillPolylines(app)
|
||||||
|
initMeilisearchDocuments(app, client)
|
||||||
|
}()
|
||||||
return nil
|
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 {
|
func initCategories(app core.App) error {
|
||||||
query := app.RecordQuery("categories")
|
query := app.RecordQuery("categories")
|
||||||
records := []*core.Record{}
|
records := []*core.Record{}
|
||||||
|
|||||||
35
db/migrations/1748000000_updated_trails_polyline.go
Normal file
35
db/migrations/1748000000_updated_trails_polyline.go
Normal file
@@ -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)
|
||||||
|
})
|
||||||
|
}
|
||||||
@@ -51,8 +51,17 @@ func RemoteListGet(e *core.RequestEvent) error {
|
|||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
updatedAt := record.GetDateTime("updated").Time()
|
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 {
|
} else {
|
||||||
|
|||||||
@@ -8,10 +8,13 @@ import (
|
|||||||
"io"
|
"io"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/url"
|
"net/url"
|
||||||
|
"os"
|
||||||
"path"
|
"path"
|
||||||
"pocketbase/federation"
|
"pocketbase/federation"
|
||||||
"pocketbase/util"
|
"pocketbase/util"
|
||||||
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/pocketbase/dbx"
|
"github.com/pocketbase/dbx"
|
||||||
@@ -19,6 +22,21 @@ import (
|
|||||||
"github.com/pocketbase/pocketbase/tools/filesystem"
|
"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 ---
|
// --- Main Handler ---
|
||||||
|
|
||||||
func RemoteTrailGet(e *core.RequestEvent) error {
|
func RemoteTrailGet(e *core.RequestEvent) error {
|
||||||
@@ -64,8 +82,17 @@ func RemoteTrailGet(e *core.RequestEvent) error {
|
|||||||
} else {
|
} else {
|
||||||
// We already have it locally. Show and update background.
|
// We already have it locally. Show and update background.
|
||||||
updatedAt := record.GetDateTime("updated").Time()
|
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 {
|
} else {
|
||||||
|
|||||||
@@ -13,8 +13,6 @@ import (
|
|||||||
|
|
||||||
"github.com/meilisearch/meilisearch-go"
|
"github.com/meilisearch/meilisearch-go"
|
||||||
"github.com/pocketbase/pocketbase/core"
|
"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) {
|
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")
|
category = trailCategory.GetString("name")
|
||||||
}
|
}
|
||||||
|
|
||||||
polyline, err := getPolyline(app, r)
|
|
||||||
if err != nil {
|
|
||||||
polyline = ""
|
|
||||||
}
|
|
||||||
|
|
||||||
domain := ""
|
domain := ""
|
||||||
if !author.GetBool("isLocal") {
|
if !author.GetBool("isLocal") {
|
||||||
domain = author.GetString("domain")
|
domain = author.GetString("domain")
|
||||||
@@ -72,7 +65,7 @@ func documentFromTrailRecord(app core.App, r *core.Record, author *core.Record,
|
|||||||
"thumbnail": thumbnail,
|
"thumbnail": thumbnail,
|
||||||
"gpx": r.GetString("gpx"),
|
"gpx": r.GetString("gpx"),
|
||||||
"tags": tags,
|
"tags": tags,
|
||||||
"polyline": polyline,
|
"polyline": r.GetString("polyline"),
|
||||||
"domain": domain,
|
"domain": domain,
|
||||||
"iri": r.GetString("iri"),
|
"iri": r.GetString("iri"),
|
||||||
"_geo": map[string]float64{
|
"_geo": map[string]float64{
|
||||||
@@ -128,46 +121,6 @@ func difficultyToNumber(difficulty string) int32 {
|
|||||||
return 0
|
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) {
|
func documentFromListRecord(r *core.Record, author *core.Record, includeShares bool) (map[string]any, error) {
|
||||||
|
|
||||||
totalElevationGain := 0.0
|
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}
|
documents := []map[string]interface{}{doc}
|
||||||
|
|
||||||
task, err := client.Index("trails").UpdateDocuments(documents, nil)
|
if _, err = client.Index("trails").UpdateDocuments(documents, nil); err != nil {
|
||||||
|
|
||||||
if err != nil {
|
|
||||||
return err
|
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
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
64
db/util/polyline.go
Normal file
64
db/util/polyline.go
Normal file
@@ -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
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user