fix(scraper): add EnsureMigrations to patch progress.user_id on startup
On every scraper startup, EnsureMigrations fetches each collection schema and PATCHes in any missing fields. This repairs the progress collection on cloud deploys where pb-init ran before user_id was added to the schema. No-ops when the field already exists.
This commit is contained in:
@@ -39,6 +39,9 @@ func NewHybridStore(ctx context.Context, pbCfg PocketBaseConfig, minioCfg MinioC
|
|||||||
// Non-fatal: 400/422 means collections already exist.
|
// Non-fatal: 400/422 means collections already exist.
|
||||||
log.Warn("EnsureCollections returned an error (may be safe to ignore)", "err", err)
|
log.Warn("EnsureCollections returned an error (may be safe to ignore)", "err", err)
|
||||||
}
|
}
|
||||||
|
if err := pb.EnsureMigrations(ctx); err != nil {
|
||||||
|
log.Warn("EnsureMigrations returned an error", "err", err)
|
||||||
|
}
|
||||||
return &HybridStore{pb: pb, minio: mc, log: log}, nil
|
return &HybridStore{pb: pb, minio: mc, log: log}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -389,6 +389,82 @@ func (s *PocketBaseStore) EnsureCollections(ctx context.Context) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ─── Schema migrations ────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
// migration describes a single field to guarantee exists in a collection.
|
||||||
|
type migration struct {
|
||||||
|
collection string
|
||||||
|
fieldName string
|
||||||
|
fieldType string
|
||||||
|
}
|
||||||
|
|
||||||
|
// migrations is the ordered list of schema changes applied on every startup.
|
||||||
|
var migrations = []migration{
|
||||||
|
// user_id was added to progress after initial deploy.
|
||||||
|
{"progress", "user_id", "text"},
|
||||||
|
}
|
||||||
|
|
||||||
|
// EnsureMigrations idempotently adds any fields that are missing from existing
|
||||||
|
// collections. It fetches the current schema, checks for each field by name,
|
||||||
|
// and PATCHes the collection only when something is absent.
|
||||||
|
// Safe to call on every startup — no-ops when schema is already up to date.
|
||||||
|
func (s *PocketBaseStore) EnsureMigrations(ctx context.Context) error {
|
||||||
|
for _, m := range migrations {
|
||||||
|
if err := s.ensureField(ctx, m); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *PocketBaseStore) ensureField(ctx context.Context, m migration) error {
|
||||||
|
// Fetch current collection schema.
|
||||||
|
resp, err := s.pb.do(ctx, http.MethodGet, "/api/collections/"+m.collection, nil)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("pocketbase: ensureField %s.%s: fetch schema: %w", m.collection, m.fieldName, err)
|
||||||
|
}
|
||||||
|
defer resp.Body.Close()
|
||||||
|
body, _ := io.ReadAll(resp.Body)
|
||||||
|
if resp.StatusCode != http.StatusOK {
|
||||||
|
return fmt.Errorf("pocketbase: ensureField %s.%s: fetch schema status %d: %s", m.collection, m.fieldName, resp.StatusCode, body)
|
||||||
|
}
|
||||||
|
|
||||||
|
var schema struct {
|
||||||
|
ID string `json:"id"`
|
||||||
|
Fields []map[string]interface{} `json:"fields"`
|
||||||
|
}
|
||||||
|
if err := json.Unmarshal(body, &schema); err != nil {
|
||||||
|
return fmt.Errorf("pocketbase: ensureField %s.%s: decode schema: %w", m.collection, m.fieldName, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Check if field already exists.
|
||||||
|
for _, f := range schema.Fields {
|
||||||
|
if name, _ := f["name"].(string); name == m.fieldName {
|
||||||
|
s.log.Debug("pocketbase: field already exists, skipping migration",
|
||||||
|
"collection", m.collection, "field", m.fieldName)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Append the new field and PATCH the collection.
|
||||||
|
newFields := append(schema.Fields, map[string]interface{}{
|
||||||
|
"name": m.fieldName,
|
||||||
|
"type": m.fieldType,
|
||||||
|
})
|
||||||
|
patch := map[string]interface{}{"fields": newFields}
|
||||||
|
patchResp, err := s.pb.do(ctx, http.MethodPatch, "/api/collections/"+schema.ID, patch)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("pocketbase: ensureField %s.%s: patch: %w", m.collection, m.fieldName, err)
|
||||||
|
}
|
||||||
|
defer patchResp.Body.Close()
|
||||||
|
patchBody, _ := io.ReadAll(patchResp.Body)
|
||||||
|
if patchResp.StatusCode != http.StatusOK {
|
||||||
|
return fmt.Errorf("pocketbase: ensureField %s.%s: patch status %d: %s", m.collection, m.fieldName, patchResp.StatusCode, patchBody)
|
||||||
|
}
|
||||||
|
s.log.Info("pocketbase: schema migration applied", "collection", m.collection, "field", m.fieldName, "type", m.fieldType)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// ─── Book metadata ────────────────────────────────────────────────────────────
|
// ─── Book metadata ────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
func (s *PocketBaseStore) UpsertBook(ctx context.Context, slug, title, author, cover, status, summary, sourceURL string, genres []string, totalChapters, ranking int) error {
|
func (s *PocketBaseStore) UpsertBook(ctx context.Context, slug, title, author, cover, status, summary, sourceURL string, genres []string, totalChapters, ranking int) error {
|
||||||
|
|||||||
Reference in New Issue
Block a user