Monorepo for Tangled
tangled.org
1package db
2
3import (
4 "context"
5 "database/sql"
6 "fmt"
7 "time"
8
9 _ "github.com/jackc/pgx/v5/stdlib"
10 "tangled.org/core/log"
11)
12
13type DBTX interface {
14 ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
15 QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
16 QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
17}
18
19func Make(ctx context.Context, dbUrl string, maxConns int) (*sql.DB, error) {
20 db, err := sql.Open("pgx", dbUrl)
21 if err != nil {
22 return nil, fmt.Errorf("opening db: %w", err)
23 }
24
25 db.SetMaxOpenConns(maxConns)
26 db.SetMaxIdleConns(maxConns)
27 db.SetConnMaxIdleTime(time.Hour)
28
29 pingCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
30 defer cancel()
31 if err := db.PingContext(pingCtx); err != nil {
32 db.Close()
33 return nil, fmt.Errorf("ping db: %w", err)
34 }
35
36 conn, err := db.Conn(ctx)
37 if err != nil {
38 return nil, err
39 }
40 defer conn.Close()
41
42 _, err = conn.ExecContext(ctx, `
43 create table if not exists repos (
44 did text not null,
45 rkey text not null,
46 at_uri text generated always as ('at://' || did || '/' || 'sh.tangled.repo' || '/' || rkey) stored,
47 cid text,
48
49 -- record content
50 name text not null,
51 knot_domain text not null,
52
53 -- sync info
54 git_rev text not null,
55 repo_sha text not null,
56 state text not null default 'pending',
57 error_msg text,
58 retry_count integer not null default 0,
59 retry_after integer not null default 0,
60 db_created_at timestamptz not null default now(),
61 db_updated_at timestamptz not null default now(),
62
63 constraint repos_pkey primary key (did, rkey)
64 );
65
66 -- knot hosts
67 create table if not exists hosts (
68 hostname text not null,
69 no_ssl boolean not null default false,
70 status text not null default 'active',
71 last_seq bigint not null default -1,
72 db_created_at timestamptz not null default now(),
73 db_updated_at timestamptz not null default now(),
74
75 constraint hosts_pkey primary key (hostname)
76 );
77
78 -- repo language stats at HEAD
79 create table if not exists repo_head_languages (
80 repo text not null, -- repo identifier (did)
81 commit text not null, -- commit id (oid)
82 language text not null,
83 size integer not null check (size >= 0),
84
85 constraint repo_head_languages_pkey
86 primary key (repo, commit, language)
87 );
88
89 create index if not exists idx_repos_aturi on repos (at_uri);
90 create index if not exists idx_repos_db_updated_at on repos (db_updated_at desc);
91 create index if not exists idx_hosts_db_updated_at on hosts (db_updated_at desc);
92
93 create index if not exists idx_repo_head_languages_repo_commit
94 on repo_head_languages (repo, commit);
95
96 create or replace function set_updated_at()
97 returns trigger as $$
98 begin
99 new.db_updated_at = now();
100 return new;
101 end;
102 $$ language plpgsql;
103
104 drop trigger if exists repos_set_updated_at on repos;
105 create trigger repos_set_updated_at
106 before update on repos
107 for each row
108 execute function set_updated_at();
109
110 drop trigger if exists hosts_set_updated_at on hosts;
111 create trigger hosts_set_updated_at
112 before update on hosts
113 for each row
114 execute function set_updated_at();
115 `)
116 if err != nil {
117 return nil, fmt.Errorf("initializing db schema: %w", err)
118 }
119
120 if err := RunMigrations(ctx, conn, log.FromContext(ctx), Migrations); err != nil {
121 return nil, fmt.Errorf("running migrations: %w", err)
122 }
123
124 return db, nil
125}