config_store.go 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371
  1. package postgres
  2. import (
  3. "context"
  4. "encoding/json"
  5. "errors"
  6. "fmt"
  7. "github.com/jackc/pgx/v5"
  8. )
  9. type EventConfigSource struct {
  10. ID string
  11. EventID string
  12. SourceVersionNo int
  13. SourceKind string
  14. SchemaID string
  15. SchemaVersion string
  16. Status string
  17. SourceJSON string
  18. Notes *string
  19. }
  20. type EventConfigBuild struct {
  21. ID string
  22. EventID string
  23. SourceID string
  24. BuildNo int
  25. BuildStatus string
  26. BuildLog *string
  27. ManifestJSON string
  28. AssetIndexJSON string
  29. }
  30. type EventReleaseAsset struct {
  31. ID string
  32. EventReleaseID string
  33. AssetType string
  34. AssetKey string
  35. AssetPath *string
  36. AssetURL string
  37. Checksum *string
  38. SizeBytes *int64
  39. MetaJSON string
  40. }
  41. type UpsertEventConfigSourceParams struct {
  42. EventID string
  43. SourceVersionNo int
  44. SourceKind string
  45. SchemaID string
  46. SchemaVersion string
  47. Status string
  48. Source map[string]any
  49. Notes *string
  50. }
  51. type UpsertEventConfigBuildParams struct {
  52. EventID string
  53. SourceID string
  54. BuildNo int
  55. BuildStatus string
  56. BuildLog *string
  57. Manifest map[string]any
  58. AssetIndex []map[string]any
  59. }
  60. type UpsertEventReleaseAssetParams struct {
  61. EventReleaseID string
  62. AssetType string
  63. AssetKey string
  64. AssetPath *string
  65. AssetURL string
  66. Checksum *string
  67. SizeBytes *int64
  68. Meta map[string]any
  69. }
  70. func (s *Store) UpsertEventConfigSource(ctx context.Context, tx Tx, params UpsertEventConfigSourceParams) (*EventConfigSource, error) {
  71. sourceJSON, err := json.Marshal(params.Source)
  72. if err != nil {
  73. return nil, fmt.Errorf("marshal event config source: %w", err)
  74. }
  75. row := tx.QueryRow(ctx, `
  76. INSERT INTO event_config_sources (
  77. event_id, source_version_no, source_kind, schema_id, schema_version, status, source_jsonb, notes
  78. )
  79. VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8)
  80. ON CONFLICT (event_id, source_version_no) DO UPDATE SET
  81. source_kind = EXCLUDED.source_kind,
  82. schema_id = EXCLUDED.schema_id,
  83. schema_version = EXCLUDED.schema_version,
  84. status = EXCLUDED.status,
  85. source_jsonb = EXCLUDED.source_jsonb,
  86. notes = EXCLUDED.notes
  87. RETURNING id, event_id, source_version_no, source_kind, schema_id, schema_version, status, source_jsonb::text, notes
  88. `, params.EventID, params.SourceVersionNo, params.SourceKind, params.SchemaID, params.SchemaVersion, params.Status, string(sourceJSON), params.Notes)
  89. var item EventConfigSource
  90. if err := row.Scan(
  91. &item.ID,
  92. &item.EventID,
  93. &item.SourceVersionNo,
  94. &item.SourceKind,
  95. &item.SchemaID,
  96. &item.SchemaVersion,
  97. &item.Status,
  98. &item.SourceJSON,
  99. &item.Notes,
  100. ); err != nil {
  101. return nil, fmt.Errorf("upsert event config source: %w", err)
  102. }
  103. return &item, nil
  104. }
  105. func (s *Store) UpsertEventConfigBuild(ctx context.Context, tx Tx, params UpsertEventConfigBuildParams) (*EventConfigBuild, error) {
  106. manifestJSON, err := json.Marshal(params.Manifest)
  107. if err != nil {
  108. return nil, fmt.Errorf("marshal event config manifest: %w", err)
  109. }
  110. assetIndexJSON, err := json.Marshal(params.AssetIndex)
  111. if err != nil {
  112. return nil, fmt.Errorf("marshal event config asset index: %w", err)
  113. }
  114. row := tx.QueryRow(ctx, `
  115. INSERT INTO event_config_builds (
  116. event_id, source_id, build_no, build_status, build_log, manifest_jsonb, asset_index_jsonb
  117. )
  118. VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7::jsonb)
  119. ON CONFLICT (event_id, build_no) DO UPDATE SET
  120. source_id = EXCLUDED.source_id,
  121. build_status = EXCLUDED.build_status,
  122. build_log = EXCLUDED.build_log,
  123. manifest_jsonb = EXCLUDED.manifest_jsonb,
  124. asset_index_jsonb = EXCLUDED.asset_index_jsonb
  125. RETURNING id, event_id, source_id, build_no, build_status, build_log, manifest_jsonb::text, asset_index_jsonb::text
  126. `, params.EventID, params.SourceID, params.BuildNo, params.BuildStatus, params.BuildLog, string(manifestJSON), string(assetIndexJSON))
  127. var item EventConfigBuild
  128. if err := row.Scan(
  129. &item.ID,
  130. &item.EventID,
  131. &item.SourceID,
  132. &item.BuildNo,
  133. &item.BuildStatus,
  134. &item.BuildLog,
  135. &item.ManifestJSON,
  136. &item.AssetIndexJSON,
  137. ); err != nil {
  138. return nil, fmt.Errorf("upsert event config build: %w", err)
  139. }
  140. return &item, nil
  141. }
  142. func (s *Store) AttachBuildToRelease(ctx context.Context, tx Tx, releaseID, buildID string) error {
  143. if _, err := tx.Exec(ctx, `
  144. UPDATE event_releases
  145. SET build_id = $2
  146. WHERE id = $1
  147. `, releaseID, buildID); err != nil {
  148. return fmt.Errorf("attach build to release: %w", err)
  149. }
  150. return nil
  151. }
  152. func (s *Store) ReplaceEventReleaseAssets(ctx context.Context, tx Tx, eventReleaseID string, assets []UpsertEventReleaseAssetParams) error {
  153. if _, err := tx.Exec(ctx, `DELETE FROM event_release_assets WHERE event_release_id = $1`, eventReleaseID); err != nil {
  154. return fmt.Errorf("clear event release assets: %w", err)
  155. }
  156. for _, asset := range assets {
  157. metaJSON, err := json.Marshal(asset.Meta)
  158. if err != nil {
  159. return fmt.Errorf("marshal event release asset meta: %w", err)
  160. }
  161. if _, err := tx.Exec(ctx, `
  162. INSERT INTO event_release_assets (
  163. event_release_id, asset_type, asset_key, asset_path, asset_url, checksum, size_bytes, meta_jsonb
  164. )
  165. VALUES ($1, $2, $3, $4, $5, $6, $7, $8::jsonb)
  166. `, eventReleaseID, asset.AssetType, asset.AssetKey, asset.AssetPath, asset.AssetURL, asset.Checksum, asset.SizeBytes, string(metaJSON)); err != nil {
  167. return fmt.Errorf("insert event release asset: %w", err)
  168. }
  169. }
  170. return nil
  171. }
  172. func (s *Store) NextEventConfigSourceVersion(ctx context.Context, eventID string) (int, error) {
  173. var next int
  174. if err := s.pool.QueryRow(ctx, `
  175. SELECT COALESCE(MAX(source_version_no), 0) + 1
  176. FROM event_config_sources
  177. WHERE event_id = $1
  178. `, eventID).Scan(&next); err != nil {
  179. return 0, fmt.Errorf("next event config source version: %w", err)
  180. }
  181. return next, nil
  182. }
  183. func (s *Store) NextEventConfigBuildNo(ctx context.Context, eventID string) (int, error) {
  184. var next int
  185. if err := s.pool.QueryRow(ctx, `
  186. SELECT COALESCE(MAX(build_no), 0) + 1
  187. FROM event_config_builds
  188. WHERE event_id = $1
  189. `, eventID).Scan(&next); err != nil {
  190. return 0, fmt.Errorf("next event config build no: %w", err)
  191. }
  192. return next, nil
  193. }
  194. func (s *Store) ListEventConfigSourcesByEventID(ctx context.Context, eventID string, limit int) ([]EventConfigSource, error) {
  195. if limit <= 0 || limit > 100 {
  196. limit = 20
  197. }
  198. rows, err := s.pool.Query(ctx, `
  199. SELECT id, event_id, source_version_no, source_kind, schema_id, schema_version, status, source_jsonb::text, notes
  200. FROM event_config_sources
  201. WHERE event_id = $1
  202. ORDER BY source_version_no DESC
  203. LIMIT $2
  204. `, eventID, limit)
  205. if err != nil {
  206. return nil, fmt.Errorf("list event config sources: %w", err)
  207. }
  208. defer rows.Close()
  209. var items []EventConfigSource
  210. for rows.Next() {
  211. item, err := scanEventConfigSourceFromRows(rows)
  212. if err != nil {
  213. return nil, err
  214. }
  215. items = append(items, *item)
  216. }
  217. if err := rows.Err(); err != nil {
  218. return nil, fmt.Errorf("iterate event config sources: %w", err)
  219. }
  220. return items, nil
  221. }
  222. func (s *Store) GetEventConfigSourceByID(ctx context.Context, sourceID string) (*EventConfigSource, error) {
  223. row := s.pool.QueryRow(ctx, `
  224. SELECT id, event_id, source_version_no, source_kind, schema_id, schema_version, status, source_jsonb::text, notes
  225. FROM event_config_sources
  226. WHERE id = $1
  227. LIMIT 1
  228. `, sourceID)
  229. return scanEventConfigSource(row)
  230. }
  231. func (s *Store) GetEventConfigBuildByID(ctx context.Context, buildID string) (*EventConfigBuild, error) {
  232. row := s.pool.QueryRow(ctx, `
  233. SELECT id, event_id, source_id, build_no, build_status, build_log, manifest_jsonb::text, asset_index_jsonb::text
  234. FROM event_config_builds
  235. WHERE id = $1
  236. LIMIT 1
  237. `, buildID)
  238. return scanEventConfigBuild(row)
  239. }
  240. func (s *Store) ListEventConfigBuildsByEventID(ctx context.Context, eventID string, limit int) ([]EventConfigBuild, error) {
  241. if limit <= 0 || limit > 100 {
  242. limit = 20
  243. }
  244. rows, err := s.pool.Query(ctx, `
  245. SELECT id, event_id, source_id, build_no, build_status, build_log, manifest_jsonb::text, asset_index_jsonb::text
  246. FROM event_config_builds
  247. WHERE event_id = $1
  248. ORDER BY build_no DESC
  249. LIMIT $2
  250. `, eventID, limit)
  251. if err != nil {
  252. return nil, fmt.Errorf("list event config builds: %w", err)
  253. }
  254. defer rows.Close()
  255. items := []EventConfigBuild{}
  256. for rows.Next() {
  257. item, err := scanEventConfigBuildFromRows(rows)
  258. if err != nil {
  259. return nil, err
  260. }
  261. items = append(items, *item)
  262. }
  263. if err := rows.Err(); err != nil {
  264. return nil, fmt.Errorf("iterate event config builds: %w", err)
  265. }
  266. return items, nil
  267. }
  268. func scanEventConfigSource(row pgx.Row) (*EventConfigSource, error) {
  269. var item EventConfigSource
  270. err := row.Scan(
  271. &item.ID,
  272. &item.EventID,
  273. &item.SourceVersionNo,
  274. &item.SourceKind,
  275. &item.SchemaID,
  276. &item.SchemaVersion,
  277. &item.Status,
  278. &item.SourceJSON,
  279. &item.Notes,
  280. )
  281. if errors.Is(err, pgx.ErrNoRows) {
  282. return nil, nil
  283. }
  284. if err != nil {
  285. return nil, fmt.Errorf("scan event config source: %w", err)
  286. }
  287. return &item, nil
  288. }
  289. func scanEventConfigSourceFromRows(rows pgx.Rows) (*EventConfigSource, error) {
  290. var item EventConfigSource
  291. if err := rows.Scan(
  292. &item.ID,
  293. &item.EventID,
  294. &item.SourceVersionNo,
  295. &item.SourceKind,
  296. &item.SchemaID,
  297. &item.SchemaVersion,
  298. &item.Status,
  299. &item.SourceJSON,
  300. &item.Notes,
  301. ); err != nil {
  302. return nil, fmt.Errorf("scan event config source row: %w", err)
  303. }
  304. return &item, nil
  305. }
  306. func scanEventConfigBuild(row pgx.Row) (*EventConfigBuild, error) {
  307. var item EventConfigBuild
  308. err := row.Scan(
  309. &item.ID,
  310. &item.EventID,
  311. &item.SourceID,
  312. &item.BuildNo,
  313. &item.BuildStatus,
  314. &item.BuildLog,
  315. &item.ManifestJSON,
  316. &item.AssetIndexJSON,
  317. )
  318. if errors.Is(err, pgx.ErrNoRows) {
  319. return nil, nil
  320. }
  321. if err != nil {
  322. return nil, fmt.Errorf("scan event config build: %w", err)
  323. }
  324. return &item, nil
  325. }
  326. func scanEventConfigBuildFromRows(rows pgx.Rows) (*EventConfigBuild, error) {
  327. var item EventConfigBuild
  328. err := rows.Scan(
  329. &item.ID,
  330. &item.EventID,
  331. &item.SourceID,
  332. &item.BuildNo,
  333. &item.BuildStatus,
  334. &item.BuildLog,
  335. &item.ManifestJSON,
  336. &item.AssetIndexJSON,
  337. )
  338. if err != nil {
  339. return nil, fmt.Errorf("scan event config build row: %w", err)
  340. }
  341. return &item, nil
  342. }