UpsertItems inserts or updates many items in a single transaction, reusing one database connection. Used by refresh, which produces hundreds of items: a per-item open/close (as UpsertItem does) is dramatically slower. Returns the number of rows successfully written.
(ctx context.Context, items []Item)
| 80 | // per-item open/close (as UpsertItem does) is dramatically slower. Returns the |
| 81 | // number of rows successfully written. |
| 82 | func (s *Store) UpsertItems(ctx context.Context, items []Item) (int, error) { |
| 83 | if len(items) == 0 { |
| 84 | return 0, nil |
| 85 | } |
| 86 | if err := s.Init(ctx); err != nil { |
| 87 | return 0, err |
| 88 | } |
| 89 | db, err := s.open() |
| 90 | if err != nil { |
| 91 | return 0, err |
| 92 | } |
| 93 | defer db.Close() |
| 94 | |
| 95 | tx, err := db.BeginTx(ctx, nil) |
| 96 | if err != nil { |
| 97 | return 0, fmt.Errorf("metadata: begin tx: %w", err) |
| 98 | } |
| 99 | stmt, err := tx.PrepareContext(ctx, ` |
| 100 | INSERT INTO metadata_items(kind, name, description, source_id, repo_owner, repo_name, repo_branch, item_path, install_key, target_apps, metadata_json, installed, installed_targets, last_seen_at, created_at, updated_at) |
| 101 | VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) |
| 102 | ON CONFLICT(kind, install_key) DO UPDATE SET |
| 103 | name=excluded.name, description=excluded.description, source_id=excluded.source_id, |
| 104 | repo_owner=excluded.repo_owner, repo_name=excluded.repo_name, repo_branch=excluded.repo_branch, |
| 105 | item_path=excluded.item_path, target_apps=excluded.target_apps, metadata_json=excluded.metadata_json, |
| 106 | last_seen_at=excluded.last_seen_at, updated_at=excluded.updated_at`) |
| 107 | if err != nil { |
| 108 | _ = tx.Rollback() |
| 109 | return 0, fmt.Errorf("metadata: prepare upsert: %w", err) |
| 110 | } |
| 111 | defer stmt.Close() |
| 112 | |
| 113 | now := timeNow() |
| 114 | written := 0 |
| 115 | for _, item := range items { |
| 116 | if _, err := stmt.ExecContext(ctx, |
| 117 | item.Kind, item.Name, item.Description, item.SourceID, |
| 118 | item.RepoOwner, item.RepoName, item.RepoBranch, item.ItemPath, |
| 119 | item.InstallKey, item.TargetApps, coalesce(item.MetadataJSON, "{}"), |
| 120 | boolToInt(item.Installed), item.InstalledTargets, now, now, now); err != nil { |
| 121 | _ = tx.Rollback() |
| 122 | return 0, fmt.Errorf("metadata: batch upsert: %w", err) |
| 123 | } |
| 124 | written++ |
| 125 | } |
| 126 | if err := tx.Commit(); err != nil { |
| 127 | return 0, fmt.Errorf("metadata: commit upsert: %w", err) |
| 128 | } |
| 129 | return written, nil |
| 130 | } |
| 131 | |
| 132 | // DeleteStale removes items of a kind whose last_seen_at is older than the given |
| 133 | // timestamp. Used after a refresh to prune resources that no longer exist |