· 8 years ago · Mar 24, 2018, 09:02 AM
1package main
2
3import (
4 "context"
5 "os"
6 "strconv"
7 "sync"
8 "time"
9
10 otgorm "github.com/echo-health/opentracing-gorm"
11 "github.com/jinzhu/gorm"
12 _ "github.com/jinzhu/gorm/dialects/postgres"
13 _ "github.com/mattes/migrate/source/file"
14 "github.com/prometheus/common/log"
15 uuid "github.com/satori/go.uuid"
16 "github.com/segmentio/ksuid"
17 "github.com/wawandco/fako"
18)
19
20var (
21 defaultConnString = "host=localhost user=postgres dbname=postgres sslmode=disable"
22 defaultMaxConn = 20
23 defaultPoolName = "default"
24 lock = &sync.Mutex{}
25 conns = map[string]*gorm.DB{}
26)
27
28const schema = `
29CREATE TABLE IF NOT EXISTS users (
30 id STRING NOT NULL,
31 firebase_id STRING NOT NULL,
32 practice_id STRING NOT NULL,
33 title STRING NOT NULL,
34 first_name STRING NOT NULL,
35 last_name STRING NOT NULL,
36 gender STRING NOT NULL,
37 date_of_birth TIMESTAMP NOT NULL,
38 create_time TIMESTAMP NULL DEFAULT now(),
39 update_time TIMESTAMP NULL DEFAULT now(),
40 CONSTRAINT "primary" PRIMARY KEY (id ASC),
41 UNIQUE INDEX idx_users_firebase_id (firebase_id ASC)
42);
43
44CREATE TABLE IF NOT EXISTS registered_addresses (
45 user_id STRING NOT NULL,
46 id STRING NOT NULL,
47 line1 STRING NOT NULL,
48 line2 STRING NULL,
49 city STRING NOT NULL,
50 postcode STRING NOT NULL,
51 country STRING NOT NULL,
52 lat DECIMAL NOT NULL,
53 lng DECIMAL NOT NULL,
54 CONSTRAINT "primary" PRIMARY KEY (user_id, id),
55 CONSTRAINT fk_user FOREIGN KEY (user_id) REFERENCES users
56) INTERLEAVE IN PARENT users (user_id);
57
58CREATE TABLE IF NOT EXISTS user_relationships (
59 user_id STRING NOT NULL,
60 account_id STRING NOT NULL,
61 relationship STRING NOT NULL,
62 create_time TIMESTAMP NULL DEFAULT now(),
63 update_time TIMESTAMP NULL DEFAULT now(),
64 CONSTRAINT "primary" PRIMARY KEY (user_id, account_id),
65 INDEX idx_users_accounts_id (account_id ASC),
66 CONSTRAINT fk_user FOREIGN KEY (user_id) REFERENCES users
67) INTERLEAVE IN PARENT users (user_id);
68
69CREATE TABLE IF NOT EXISTS user_exemptions (
70 id STRING NOT NULL,
71 user_id STRING NOT NULL,
72 code STRING NOT NULL,
73 status STRING NOT NULL,
74 url STRING NULL,
75 expiry TIMESTAMP NULL DEFAULT now(),
76 create_time TIMESTAMP NULL DEFAULT now(),
77 update_time TIMESTAMP NULL DEFAULT now(),
78 CONSTRAINT "primary" PRIMARY KEY (user_id, id),
79 CONSTRAINT fk_user FOREIGN KEY (user_id) REFERENCES users
80) INTERLEAVE IN PARENT users (user_id);
81`
82
83type User struct {
84 ID string
85 PracticeID string
86 FirebaseID string
87 Title string `fako:"title"`
88 FirstName string `fako:"first_name"`
89 LastName string `fako:"last_name"`
90 Gender string
91 DateOfBirth time.Time
92 CreatedAt time.Time `gorm:"column:create_time"`
93 UpdatedAt time.Time `gorm:"column:update_time"`
94 Relationships []*UserRelationship
95 Exemptions []*UserExemption
96 RegisteredAddress RegisteredAddress
97}
98
99type UserRelationship struct {
100 UserID string `gorm:"primary_key"`
101 AccountID string `gorm:"primary_key" fako:"simple_password"`
102 Relationship string `fako:"first_name"`
103 CreatedAt time.Time `gorm:"column:create_time"`
104 UpdatedAt time.Time `gorm:"column:update_time"`
105}
106
107type RegisteredAddress struct {
108 ID string
109 UserID string
110 Line1 string `fako:"street_address"`
111 Line2 string `fako:"street"`
112 City string `fako:"city"`
113 Postcode string `fako:"zip"`
114 Country string `fako:"country"`
115 Lat float64
116 Lng float64
117}
118
119type UserExemption struct {
120 ID string
121 UserID string
122 URL string `fako:"characters"`
123 Code string
124 Status string
125 Expiry *time.Time
126 CreatedAt time.Time `gorm:"column:create_time"`
127 UpdatedAt time.Time `gorm:"column:update_time"`
128}
129
130func UserID() string {
131 if os.Getenv("USE_UUID") != "" {
132 u, _ := uuid.NewV4()
133 return u.String()
134 }
135
136 return "user_" + ksuid.New().String()
137}
138
139func ExemptionID() string {
140 if os.Getenv("USE_UUID") != "" {
141 u, _ := uuid.NewV4()
142 return u.String()
143 }
144
145 return "user_ex_" + ksuid.New().String()
146}
147
148func NominationRequestID() string {
149 if os.Getenv("USE_UUID") != "" {
150 u, _ := uuid.NewV4()
151 return u.String()
152 }
153
154 return "user_nom_" + ksuid.New().String()
155}
156
157func AddrID() string {
158 if os.Getenv("USE_UUID") != "" {
159 u, _ := uuid.NewV4()
160 return u.String()
161 }
162
163 return "addr_" + ksuid.New().String()
164}
165
166func (p *User) BeforeCreate(scope *gorm.Scope) error {
167 if p.ID == "" {
168 p.ID = UserID()
169 scope.SetColumn("ID", p.ID)
170 }
171
172 if p.FirebaseID == "" {
173 p.FirebaseID = p.ID
174 }
175
176 return nil
177}
178
179func (p *RegisteredAddress) BeforeCreate(scope *gorm.Scope) error {
180 if p.ID == "" {
181 p.ID = AddrID()
182 return scope.SetColumn("ID", p.ID)
183 }
184
185 return nil
186}
187
188func (p *UserExemption) BeforeCreate(scope *gorm.Scope) error {
189 if p.ID == "" {
190 p.ID = ExemptionID()
191 return scope.SetColumn("ID", p.ID)
192 }
193
194 return nil
195}
196
197func GetDB(ctx context.Context) (*gorm.DB, error) {
198 return GetDBFromPool(ctx, defaultPoolName)
199}
200
201func GetDBFromPool(ctx context.Context, pool string) (*gorm.DB, error) {
202 lock.Lock()
203 defer lock.Unlock()
204
205 if conns[pool] != nil {
206 db := otgorm.SetSpanToGorm(ctx, conns[pool])
207 return db, nil
208 }
209
210 connString := defaultConnString
211 if os.Getenv("POSTGRES_URL") != "" {
212 connString = os.Getenv("POSTGRES_URL")
213 }
214
215 log.Debugf("Connecting to PG at %s for pool %s", connString, pool)
216 var err error
217 conn, err := connect(connString)
218 if err != nil {
219 db := otgorm.SetSpanToGorm(ctx, conn)
220 return db, err
221 }
222
223 conns[pool] = conn
224 return conns[pool], err
225}
226
227func connect(connString string) (*gorm.DB, error) {
228 driver := "postgres"
229 if os.Getenv("CLOUDSQL") == "true" {
230 driver = "cloudsqlpostgres"
231 }
232
233 conn, err := gorm.Open(driver, connString)
234
235 if err != nil {
236 return nil, err
237 }
238
239 maxConns := defaultMaxConn
240 if os.Getenv("DB_MAX_CONNS") != "" {
241 i, err := strconv.Atoi(os.Getenv("DB_MAX_CONNS"))
242 if err != nil {
243 return nil, err
244 }
245
246 maxConns = i
247 }
248
249 conn.DB().SetMaxOpenConns(maxConns)
250
251 if os.Getenv("DEBUG_SQL") == "true" {
252 conn.LogMode(true)
253 }
254
255 otgorm.AddGormCallbacks(conn)
256
257 conn = conn.Set("gorm:auto_preload", true)
258 return conn, nil
259}
260
261func Get(ctx context.Context, id string) (*User, error) {
262 db, err := GetDB(ctx)
263 if err != nil {
264 return nil, err
265 }
266
267 var p User
268 err = db.Where("id = ? OR firebase_id = ?", id, id).
269 First(&p).Error
270 return &p, err
271}
272
273func Create(ctx context.Context, new *User) error {
274 db, err := GetDBFromPool(ctx, "create")
275 if err != nil {
276 return err
277 }
278
279 old, err := Get(ctx, new.FirebaseID)
280 if err == gorm.ErrRecordNotFound {
281 return db.Create(new).Error
282 }
283
284 if err != nil {
285 return err
286 }
287
288 tx := db.Begin()
289
290 // first delete any relationships that have been removed from firebase
291 for _, rel := range old.Relationships {
292 inFirebase := false
293 for _, a := range new.Relationships {
294 if a.AccountID == rel.AccountID {
295 inFirebase = true
296 }
297 }
298
299 if !inFirebase {
300 if err = tx.Delete(rel).Error; err != nil {
301 tx.Rollback()
302 return err
303 }
304 }
305 }
306
307 new.ID = old.ID
308 new.RegisteredAddress.ID = old.RegisteredAddress.ID
309 err = tx.Save(&new).Error
310 if err != nil {
311 tx.Rollback()
312 return err
313 }
314
315 return tx.Commit().Error
316}
317
318func worker(id int, jobs <-chan *User) {
319 log.Infof("Booting worker %d", id)
320 for j := range jobs {
321 err := Create(context.Background(), j)
322 if err != nil {
323 log.Fatal(err)
324 }
325 }
326}
327
328func main() {
329 db, err := GetDB(context.Background())
330 if err != nil {
331 log.Fatal(err)
332 }
333
334 err = db.Exec(schema).Error
335 if err != nil {
336 log.Fatal(err)
337 }
338
339 jobs := make(chan *User, 100)
340
341 for w := 1; w <= 20; w++ {
342 go worker(w, jobs)
343 }
344
345 for i := 0; i < 10000; i++ {
346 u := &User{}
347 fako.Fill(u)
348
349 ra := &RegisteredAddress{}
350 fako.Fill(ra)
351 u.RegisteredAddress = *ra
352
353 for i := 0; i < 2; i++ {
354 r := &UserRelationship{}
355 fako.Fill(r)
356 u.Relationships = append(u.Relationships, r)
357 }
358
359 for i := 0; i < 2; i++ {
360 r := &UserExemption{}
361 fako.Fill(r)
362 u.Exemptions = append(u.Exemptions, r)
363 }
364
365 jobs <- u
366 }
367}