1 package sources
2
3
4
5
6 import (
7 "context"
8 "errors"
9
10 "github.com/cybertec-postgresql/pgwatch/v6/internal/db"
11 "github.com/cybertec-postgresql/pgwatch/v6/internal/metrics"
12 pgx "github.com/jackc/pgx/v5"
13 "github.com/jackc/pgx/v5/pgconn"
14 )
15
16 func NewPostgresSourcesReaderWriter(ctx context.Context, connstr string) (ReaderWriter, error) {
17 conn, err := db.New(ctx, connstr)
18 if err != nil {
19 return nil, err
20 }
21 return NewPostgresSourcesReaderWriterConn(ctx, conn)
22 }
23
24 func NewPostgresSourcesReaderWriterConn(ctx context.Context, conn db.PgxPoolIface) (ReaderWriter, error) {
25 if err := metrics.EnsureConfigSchema(ctx, conn); err != nil {
26 return nil, err
27 }
28 r := &dbSourcesReaderWriter{
29 ctx: ctx,
30 configDb: conn,
31 }
32 return r, conn.Ping(ctx)
33 }
34
35 type dbSourcesReaderWriter struct {
36 ctx context.Context
37 configDb db.PgxIface
38 }
39
40
41 var _ db.Migrator = (*dbSourcesReaderWriter)(nil)
42
43 func (r *dbSourcesReaderWriter) Migrate() error {
44 return metrics.MigrateConfigSchema(r.ctx, r.configDb)
45 }
46
47 func (r *dbSourcesReaderWriter) NeedsMigration() (bool, error) {
48 return metrics.NeedsConfigSchemaMigration(r.ctx, r.configDb)
49 }
50
51 func (r *dbSourcesReaderWriter) WriteSources(dbs Sources) error {
52 tx, err := r.configDb.Begin(context.Background())
53 if err != nil {
54 return err
55 }
56 if _, err = tx.Exec(context.Background(), `truncate pgwatch.source`); err != nil {
57 return err
58 }
59 defer func() { _ = tx.Rollback(context.Background()) }()
60 for _, md := range dbs {
61 if err = r.updateSource(tx, md); err != nil {
62 return err
63 }
64 }
65 return tx.Commit(context.Background())
66 }
67
68 func (r *dbSourcesReaderWriter) updateSource(conn db.PgxIface, md Source) (err error) {
69 m := db.MarshallParamToJSONB
70 sql := `insert into pgwatch.source(
71 name,
72 "group",
73 dbtype,
74 connstr,
75 config,
76 config_standby,
77 preset_config,
78 preset_config_standby,
79 include_pattern,
80 exclude_pattern,
81 custom_tags,
82 only_if_master,
83 is_enabled)
84 values
85 ($1, $2, $3, $4, $5, $6, NULLIF($7, ''), NULLIF($8, ''), $9, $10, $11, $12, $13)
86 on conflict (name) do update set
87 "group" = $2,
88 dbtype = $3,
89 connstr = $4,
90 config = $5,
91 config_standby = $6,
92 preset_config = NULLIF($7, ''),
93 preset_config_standby = NULLIF($8, ''),
94 include_pattern = $9,
95 exclude_pattern = $10,
96 custom_tags = $11,
97 only_if_master = $12,
98 is_enabled = $13`
99 _, err = conn.Exec(context.Background(), sql,
100 md.Name, md.Group, md.Kind,
101 md.ConnStr, m(md.Metrics), m(md.MetricsStandby), md.PresetMetrics, md.PresetMetricsStandby,
102 md.IncludePattern, md.ExcludePattern, m(md.CustomTags),
103 md.OnlyIfMaster, md.IsEnabled)
104 return err
105 }
106
107 func (r *dbSourcesReaderWriter) createSource(conn db.PgxIface, md Source) (err error) {
108 m := db.MarshallParamToJSONB
109 sql := `insert into pgwatch.source(
110 name,
111 "group",
112 dbtype,
113 connstr,
114 config,
115 config_standby,
116 preset_config,
117 preset_config_standby,
118 include_pattern,
119 exclude_pattern,
120 custom_tags,
121 only_if_master,
122 is_enabled)
123 values
124 ($1, $2, $3, $4, $5, $6, NULLIF($7, ''), NULLIF($8, ''), $9, $10, $11, $12, $13)`
125 _, err = conn.Exec(context.Background(), sql,
126 md.Name, md.Group, md.Kind,
127 md.ConnStr, m(md.Metrics), m(md.MetricsStandby), md.PresetMetrics, md.PresetMetricsStandby,
128 md.IncludePattern, md.ExcludePattern, m(md.CustomTags),
129 md.OnlyIfMaster, md.IsEnabled)
130 if err != nil {
131
132 var pgErr *pgconn.PgError
133 if errors.As(err, &pgErr) && pgErr.SQLState() == "23505" {
134 return ErrSourceExists
135 }
136 }
137 return err
138 }
139
140 func (r *dbSourcesReaderWriter) UpdateSource(md Source) error {
141 return r.updateSource(r.configDb, md)
142 }
143
144 func (r *dbSourcesReaderWriter) CreateSource(md Source) error {
145 return r.createSource(r.configDb, md)
146 }
147
148 func (r *dbSourcesReaderWriter) DeleteSource(name string) error {
149 _, err := r.configDb.Exec(context.Background(), `delete from pgwatch.source where name = $1`, name)
150 return err
151 }
152
153 func (r *dbSourcesReaderWriter) GetSources() (Sources, error) {
154 sqlLatest := `select /* pgwatch_generated */
155 name,
156 "group",
157 dbtype,
158 connstr,
159 coalesce(config, '{}'::jsonb) as config,
160 coalesce(config_standby, '{}'::jsonb) as config_standby,
161 coalesce(preset_config, '') as preset_config,
162 coalesce(preset_config_standby, '') as preset_config_standby,
163 coalesce(include_pattern, '') as include_pattern,
164 coalesce(exclude_pattern, '') as exclude_pattern,
165 coalesce(custom_tags, '{}'::jsonb) as custom_tags,
166 only_if_master,
167 is_enabled
168 from
169 pgwatch.source`
170 rows, err := r.configDb.Query(context.Background(), sqlLatest)
171 if err != nil {
172 return nil, err
173 }
174 return pgx.CollectRows[Source](rows, pgx.RowToStructByNameLax)
175 }
176