// This program demonstrates a serializability violation.
// Before running, start a postgres DB on port 5432, as with
//   docker run -d -p 5432:5432 -e LANG=C -e POSTGRES_PASSWORD=pwd postgres:13.2 

package main

import (
    "context"
    "database/sql"
    "errors"
    "fmt"
    "log"

    "github.com/lib/pq"
)

func main() {
    ctx := context.Background()

    uri := fmt.Sprintf("postgres://127.0.0.1/postgres?sslmode=disable&user=postgres&password=pwd&port=5432")
    db, err := sql.Open("postgres", uri)
    if err != nil {
        log.Fatal(err)
    }
    defer db.Close()

    _, err = db.ExecContext(ctx, `                                                                                                                                                                                                            
            DROP TABLE IF EXISTS paths;                                                                                                                                                                                                       
            CREATE TABLE  paths (                                                                                                                                                                                                             
                id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,                                                                                                                                                                          
                path TEXT NOT NULL                                                                                                                                                                                                            
                                                                                                                                                                                                                                              
            );                                                                                                                                                                                                                                
            ALTER TABLE paths ADD CONSTRAINT paths_path_key UNIQUE (path);                                                                                                                                                                    
            CREATE INDEX idx_paths_path_id ON paths(path, id);                                                                                                                                                                                
                                                                                                                                                                                                                                              
`)
    if err != nil {
        log.Fatal(err)
    }

    const n = 10
    errc := make(chan error, n)
    for i := 0; i < n; i++ {
        go func() { errc <- upsertPathInTx(ctx, db, "example.com/retractions") }()
    }
    for i := 0; i < n; i++ {
        if err := <-errc; err != nil {
            log.Fatal(err)
        }
    }
}

func upsertPathInTx(ctx context.Context, db *sql.DB, path string) error {
    for i := 0; i < 10; i++ {
        tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
        if err != nil {
            return err
        }

        var id int
        err = tx.QueryRowContext(ctx,
            `SELECT id FROM paths WHERE path = $1`,
            path).Scan(&id)
        if err == sql.ErrNoRows {
            err = tx.QueryRowContext(ctx,
                `INSERT INTO paths (path) VALUES ($1) RETURNING id`,
                path).Scan(&id)
        }
        if err != nil {
            tx.Rollback()
            if isSerializationFailure(err) {
                continue
            }
            return err
        }
        tx.Commit()
        return nil
    }
    return errors.New("too many retries")
}

const serializationFailureCode = "40001"

func isSerializationFailure(err error) (b bool) {
    var perr *pq.Error
    return errors.As(err, &perr) && perr.Code == serializationFailureCode
}