appview/db: transactions all the things

appview/db/db.go

···
 )
 
 type DB struct {
-	db *sql.DB
+	Db *sql.DB
 }
 
 func Make(dbPath string) (*DB, error) {
···
 	if err != nil {
 		return nil, err
 	}
-	return &DB{db: db}, nil
+	return &DB{Db: db}, nil
 }

appview/db/pubkeys.go

···
 package db
 
 import (
+	"database/sql"
 	"encoding/json"
 	"time"
 )
 
 func (d *DB) AddPublicKey(did, name, key string) error {
 	query := `insert into public_keys (did, name, key) values (?, ?, ?)`
-	_, err := d.db.Exec(query, did, name, key)
+	_, err := d.Db.Exec(query, did, name, key)
+	return err
+}
+
+func (d *DB) AddPublicKeyTx(tx *sql.Tx, did, name, key string) error {
+	query := `insert into public_keys (did, name, key) values (?, ?, ?)`
+	_, err := tx.Exec(query, did, name, key)
 	return err
 }
 
 func (d *DB) RemovePublicKey(did string) error {
 	query := `delete from public_keys where did = ?`
-	_, err := d.db.Exec(query, did)
+	_, err := d.Db.Exec(query, did)
 	return err
 }
 
···
 func (d *DB) GetAllPublicKeys() ([]PublicKey, error) {
 	var keys []PublicKey
 
-	rows, err := d.db.Query(`select key, name, did, created from public_keys`)
+	rows, err := d.Db.Query(`select key, name, did, created from public_keys`)
 	if err != nil {
 		return nil, err
 	}
···
 func (d *DB) GetPublicKeys(did string) ([]PublicKey, error) {
 	var keys []PublicKey
 
-	rows, err := d.db.Query(`select did, key, name, created from public_keys where did = ?`, did)
+	rows, err := d.Db.Query(`select did, key, name, created from public_keys where did = ?`, did)
 	if err != nil {
 		return nil, err
 	}

appview/db/registration.go

···
 func (d *DB) RegistrationsByDid(did string) ([]Registration, error) {
 	var registrations []Registration
 
-	rows, err := d.db.Query(`
+	rows, err := d.Db.Query(`
 		select domain, did, created, registered from registrations
 		where did = ?
 	`, did)
···
 	var registeredAt *int64
 	var registration Registration
 
-	err := d.db.QueryRow(`
+	err := d.Db.QueryRow(`
+		select domain, did, created, registered from registrations
+		where domain = ?
+	`, domain).Scan(&registration.Domain, &registration.ByDid, &createdAt, &registeredAt)
+
+	if err != nil {
+		if err == sql.ErrNoRows {
+			return nil, nil
+		} else {
+			return nil, err
+		}
+	}
+
+	createdAtTime := time.Unix(*createdAt, 0)
+	var registeredAtTime *time.Time
+	if registeredAt != nil {
+		x := time.Unix(*registeredAt, 0)
+		registeredAtTime = &x
+	}
+
+	registration.Created = &createdAtTime
+	registration.Registered = registeredAtTime
+
+	return &registration, nil
+}
+
+func (d *DB) RegistrationByDomainTx(tx *sql.Tx, domain string) (*Registration, error) {
+	var createdAt *int64
+	var registeredAt *int64
+	var registration Registration
+
+	err := tx.QueryRow(`
 		select domain, did, created, registered from registrations
 		where domain = ?
 	`, domain).Scan(&registration.Domain, &registration.ByDid, &createdAt, &registeredAt)
···
 
 	secret := uuid.New().String()
 
-	_, err = d.db.Exec(`
+	_, err = d.Db.Exec(`
 		insert into registrations (domain, did, secret)
 		values (?, ?, ?)
 		on conflict(domain) do update set did = excluded.did, secret = excluded.secret
···
 }
 
 func (d *DB) GetRegistrationKey(domain string) (string, error) {
-	res := d.db.QueryRow(`select secret from registrations where domain = ?`, domain)
+	res := d.Db.QueryRow(`select secret from registrations where domain = ?`, domain)
+
+	var secret string
+	err := res.Scan(&secret)
+	if err != nil || secret == "" {
+		return "", err
+	}
+
+	return secret, nil
+}
+
+func (d *DB) GetRegistrationKeyTx(tx *sql.Tx, domain string) (string, error) {
+	res := tx.QueryRow(`select secret from registrations where domain = ?`, domain)
 
 	var secret string
 	err := res.Scan(&secret)
···
 }
 
 func (d *DB) Register(domain string) error {
-	_, err := d.db.Exec(`
+	_, err := d.Db.Exec(`
+		update registrations
+		set registered = strftime('%s', 'now')
+		where domain = ?;
+		`, domain)
+
+	return err
+}
+
+func (d *DB) RegisterTx(tx *sql.Tx, domain string) error {
+	_, err := tx.Exec(`
 		update registrations
 		set registered = strftime('%s', 'now')
 		where domain = ?;

appview/db/repos.go

···
 package db
 
+import "database/sql"
+
 type Repo struct {
 	Did     string
 	Name    string
···
 func (d *DB) GetAllReposByDid(did string) ([]Repo, error) {
 	var repos []Repo
 
-	rows, err := d.db.Query(`select did, name, knot, created from repos where did = ?`, did)
+	rows, err := d.Db.Query(`select did, name, knot, created from repos where did = ?`, did)
 	if err != nil {
 		return nil, err
 	}
···
 func (d *DB) GetRepo(did, name string) (*Repo, error) {
 	var repo Repo
 
-	row := d.db.QueryRow(`select did, name, knot, created from repos where did = ? and name = ?`, did, name)
+	row := d.Db.QueryRow(`select did, name, knot, created from repos where did = ? and name = ?`, did, name)
 	var createdAt *int64
 	if err := row.Scan(&repo.Did, &repo.Name, &repo.Knot, &createdAt); err != nil {
 		return nil, err
···
 }
 
 func (d *DB) AddRepo(repo *Repo) error {
-	_, err := d.db.Exec(`insert into repos (did, name, knot) values (?, ?, ?)`, repo.Did, repo.Name, repo.Knot)
+	_, err := d.Db.Exec(`insert into repos (did, name, knot) values (?, ?, ?)`, repo.Did, repo.Name, repo.Knot)
+	return err
+}
+
+func (d *DB) AddRepoTx(tx *sql.Tx, repo *Repo) error {
+	_, err := tx.Exec(`insert into repos (did, name, knot) values (?, ?, ?)`, repo.Did, repo.Name, repo.Knot)
 	return err
 }
 
 func (d *DB) RemoveRepo(did, name, knot string) error {
-	_, err := d.db.Exec(`delete from repos where did = ? and name = ? and knot = ?`, did, name, knot)
+	_, err := d.Db.Exec(`delete from repos where did = ? and name = ? and knot = ?`, did, name, knot)
 	return err
 }

appview/state/state.go

···
 			return
 		}
 
-		if err := s.db.AddPublicKey(did, name, key); err != nil {
+		// Start transaction
+		tx, err := s.db.Db.Begin()
+		if err != nil {
+			log.Printf("failed to start transaction: %s", err)
+			http.Error(w, "Internal server error", http.StatusInternalServerError)
+			return
+		}
+		defer tx.Rollback() // Will rollback if not committed
+
+		if err := s.db.AddPublicKeyTx(tx, did, name, key); err != nil {
 			log.Printf("adding public key: %s", err)
 			return
 		}
···
 			return
 		}
 
+		// If everything succeeded, commit the transaction
+		if err := tx.Commit(); err != nil {
+			log.Printf("failed to commit transaction: %s", err)
+			http.Error(w, "Internal server error", http.StatusInternalServerError)
+			return
+		}
+
 		log.Println("created atproto record: ", resp.Uri)
 
 		return
···
 	}
 	log.Println("checking ", domain)
 
-	secret, err := s.db.GetRegistrationKey(domain)
+	// Start transaction
+	tx, err := s.db.Db.Begin()
+	if err != nil {
+		log.Printf("failed to start transaction: %s", err)
+		http.Error(w, "Internal server error", http.StatusInternalServerError)
+		return
+	}
+	defer tx.Rollback() // Will rollback if not committed
+
+	secret, err := s.db.GetRegistrationKeyTx(tx, domain)
 	if err != nil {
 		log.Printf("no key found for domain %s: %s\n", domain, err)
 		return
···
 		return
 	}
 
-	// mark as registered
-	err = s.db.Register(domain)
+	// mark as registered within transaction
+	err = s.db.RegisterTx(tx, domain)
 	if err != nil {
 		log.Println("failed to register domain", err)
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
 
-	// set permissions for this did as owner
-	reg, err := s.db.RegistrationByDomain(domain)
+	// set permissions for this did as owner within transaction
+	reg, err := s.db.RegistrationByDomainTx(tx, domain)
 	if err != nil {
 		log.Println("failed to register domain", err)
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
 
-	// add basic acls for this domain
+	// add basic acls for this domain within transaction
 	err = s.enforcer.AddDomain(domain)
 	if err != nil {
 		log.Println("failed to setup owner of domain", err)
···
 		return
 	}
 
-	// add this did as owner of this domain
+	// add this did as owner of this domain within transaction
 	err = s.enforcer.AddOwner(domain, reg.ByDid)
 	if err != nil {
 		log.Println("failed to setup owner of domain", err)
 		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	// Commit transaction
+	if err := tx.Commit(); err != nil {
+		log.Printf("failed to commit transaction: %s", err)
+		http.Error(w, "Internal server error", http.StatusInternalServerError)
 		return
 	}
 
···
 	}
 	log.Printf("adding %s to %s\n", memberIdent.Handle.String(), domain)
 
-	// announce this relation into the firehose, store into owners' pds
-	client, _ := s.auth.AuthorizedClient(r)
-	currentUser := s.auth.GetUser(r)
-	addedAt := time.Now().Format(time.RFC3339)
-	resp, err := comatproto.RepoPutRecord(r.Context(), client, &comatproto.RepoPutRecord_Input{
-		Collection: tangled.KnotMemberNSID,
-		Repo:       currentUser.Did,
-		Rkey:       s.TID(),
-		Record: &lexutil.LexiconTypeDecoder{
-			Val: &tangled.KnotMember{
-				Member:  memberIdent.DID.String(),
-				Domain:  domain,
-				AddedAt: &addedAt,
-			}},
-	})
-	// invalid record
+	// Start transaction
+	tx, err := s.db.Db.Begin()
 	if err != nil {
-		log.Printf("failed to create record: %s", err)
+		log.Printf("failed to start transaction: %s", err)
+		http.Error(w, "Internal server error", http.StatusInternalServerError)
 		return
 	}
-	log.Println("created atproto record: ", resp.Uri)
+	defer tx.Rollback() // Will rollback if not committed
 
-	secret, err := s.db.GetRegistrationKey(domain)
+	// Get registration key within transaction
+	secret, err := s.db.GetRegistrationKeyTx(tx, domain)
 	if err != nil {
 		log.Printf("no key found for domain %s: %s\n", domain, err)
 		return
 	}
 
+	// Make the external call to the knot server
 	ksClient, err := NewSignedClient(domain, secret)
 	if err != nil {
 		log.Println("failed to create client to ", domain)
···
 
 	ksResp, err := ksClient.AddMember(memberIdent.DID.String(), []string{})
 	if err != nil {
-		log.Printf("failet to make request to %s: %s", domain, err)
+		log.Printf("failed to make request to %s: %s", domain, err)
+		return
 	}
 
 	if ksResp.StatusCode != http.StatusNoContent {
···
 		return
 	}
 
+	// Create ATProto record within transaction
+	client, _ := s.auth.AuthorizedClient(r)
+	currentUser := s.auth.GetUser(r)
+	addedAt := time.Now().Format(time.RFC3339)
+	resp, err := comatproto.RepoPutRecord(r.Context(), client, &comatproto.RepoPutRecord_Input{
+		Collection: tangled.KnotMemberNSID,
+		Repo:       currentUser.Did,
+		Rkey:       s.TID(),
+		Record: &lexutil.LexiconTypeDecoder{
+			Val: &tangled.KnotMember{
+				Member:  memberIdent.DID.String(),
+				Domain:  domain,
+				AddedAt: &addedAt,
+			}},
+	})
+	if err != nil {
+		log.Printf("failed to create record: %s", err)
+		return
+	}
+
+	// Update RBAC within transaction
 	err = s.enforcer.AddMember(domain, memberIdent.DID.String())
 	if err != nil {
 		w.Write([]byte(fmt.Sprint("failed to add member: ", err)))
 		return
 	}
 
+	// If everything succeeded, commit the transaction
+	if err := tx.Commit(); err != nil {
+		log.Printf("failed to commit transaction: %s", err)
+		http.Error(w, "Internal server error", http.StatusInternalServerError)
+		return
+	}
+
+	log.Println("created atproto record: ", resp.Uri)
 	w.Write([]byte(fmt.Sprint("added member: ", memberIdent.Handle.String())))
 }
 
···
 			return
 		}
 
-		secret, err := s.db.GetRegistrationKey(domain)
+		// Start transaction
+		tx, err := s.db.Db.Begin()
+		if err != nil {
+			log.Printf("failed to start transaction: %s", err)
+			http.Error(w, "Internal server error", http.StatusInternalServerError)
+			return
+		}
+		defer tx.Rollback() // Will rollback if not committed
+
+		secret, err := s.db.GetRegistrationKeyTx(tx, domain)
 		if err != nil {
 			log.Printf("no key found for domain %s: %s\n", domain, err)
 			return
···
 		client, err := NewSignedClient(domain, secret)
 		if err != nil {
 			log.Println("failed to create client to ", domain)
+			return
 		}
 
 		resp, err := client.NewRepo(user.Did, repoName)
···
 			return
 		}
 
-		// add to local db
+		// add to local db within transaction
 		repo := &db.Repo{
 			Did:  user.Did,
 			Name: repoName,
 			Knot: domain,
 		}
-		err = s.db.AddRepo(repo)
+		err = s.db.AddRepoTx(tx, repo)
 		if err != nil {
 			log.Println("failed to add repo to db", err)
 			return
 		}
 
-		// acls
+		// acls within transaction
 		err = s.enforcer.AddRepo(user.Did, domain, filepath.Join(user.Did, repoName))
 		if err != nil {
 			log.Println("failed to set up acls", err)
+			return
+		}
+
+		// Commit transaction
+		if err := tx.Commit(); err != nil {
+			log.Printf("failed to commit transaction: %s", err)
+			http.Error(w, "Internal server error", http.StatusInternalServerError)
 			return
 		}