From 6cf863b3969ffc663c6550bda931f04c7aa19cf8 Mon Sep 17 00:00:00 2001 From: hailey Date: Sun, 8 Jun 2025 15:56:09 -0700 Subject: [PATCH] create a db wrapper that allows for easy locking (#13) --- blockstore/blockstore.go | 18 +-- cmd/cocoon/main.go | 32 ++++ go.mod | 23 +-- go.sum | 28 +--- internal/db/db.go | 65 ++++++++ server/common.go | 4 +- server/handle_actor_put_preferences.go | 2 +- server/handle_identity_update_handle.go | 2 +- server/handle_import_repo.go | 2 +- server/handle_repo_describe_repo.go | 2 +- server/handle_repo_get_record.go | 2 +- server/handle_repo_list_records.go | 2 +- server/handle_repo_list_repos.go | 2 +- server/handle_repo_upload_blob.go | 6 +- server/handle_server_confirm_email.go | 2 +- server/handle_server_create_account.go | 8 +- server/handle_server_create_invite_code.go | 2 +- server/handle_server_create_invite_codes.go | 2 +- server/handle_server_create_session.go | 6 +- server/handle_server_delete_session.go | 4 +- server/handle_server_refresh_session.go | 4 +- ...andle_server_request_email_confirmation.go | 2 +- server/handle_server_request_email_update.go | 2 +- .../handle_server_request_password_reset.go | 2 +- server/handle_server_reset_password.go | 2 +- server/handle_server_update_email.go | 2 +- server/handle_sync_get_blob.go | 6 +- server/handle_sync_get_record.go | 2 +- server/handle_sync_get_repo.go | 2 +- server/handle_sync_list_blobs.go | 2 +- server/repo.go | 20 +-- server/server.go | 152 +++++++++++++++++- server/session.go | 4 +- 33 files changed, 317 insertions(+), 99 deletions(-) create mode 100644 internal/db/db.go diff --git a/blockstore/blockstore.go b/blockstore/blockstore.go index 06929c3..2dee9ff 100644 --- a/blockstore/blockstore.go +++ b/blockstore/blockstore.go @@ -5,21 +5,21 @@ import ( "fmt" "github.com/bluesky-social/indigo/atproto/syntax" + "github.com/haileyok/cocoon/internal/db" "github.com/haileyok/cocoon/models" blocks "github.com/ipfs/go-block-format" "github.com/ipfs/go-cid" - "gorm.io/gorm" "gorm.io/gorm/clause" ) type SqliteBlockstore struct { - db *gorm.DB + db *db.DB did string readonly bool inserts map[cid.Cid]blocks.Block } -func New(did string, db *gorm.DB) *SqliteBlockstore { +func New(did string, db *db.DB) *SqliteBlockstore { return &SqliteBlockstore{ did: did, db: db, @@ -28,7 +28,7 @@ func New(did string, db *gorm.DB) *SqliteBlockstore { } } -func NewReadOnly(did string, db *gorm.DB) *SqliteBlockstore { +func NewReadOnly(did string, db *db.DB) *SqliteBlockstore { return &SqliteBlockstore{ did: did, db: db, @@ -45,7 +45,7 @@ func (bs *SqliteBlockstore) Get(ctx context.Context, cid cid.Cid) (blocks.Block, return maybeBlock, nil } - if err := bs.db.Raw("SELECT * FROM blocks WHERE did = ? AND cid = ?", bs.did, cid.Bytes()).Scan(&block).Error; err != nil { + if err := bs.db.Raw("SELECT * FROM blocks WHERE did = ? AND cid = ?", nil, bs.did, cid.Bytes()).Scan(&block).Error; err != nil { return nil, err } @@ -71,10 +71,10 @@ func (bs *SqliteBlockstore) Put(ctx context.Context, block blocks.Block) error { Value: block.RawData(), } - if err := bs.db.Clauses(clause.OnConflict{ + if err := bs.db.Create(&b, []clause.Expression{clause.OnConflict{ Columns: []clause.Column{{Name: "did"}, {Name: "cid"}}, UpdateAll: true, - }).Create(&b).Error; err != nil { + }}).Error; err != nil { return err } @@ -94,7 +94,7 @@ func (bs *SqliteBlockstore) GetSize(context.Context, cid.Cid) (int, error) { } func (bs *SqliteBlockstore) PutMany(ctx context.Context, blocks []blocks.Block) error { - tx := bs.db.Begin() + tx := bs.db.BeginDangerously() for _, block := range blocks { bs.inserts[block.Cid()] = block @@ -137,7 +137,7 @@ func (bs *SqliteBlockstore) HashOnRead(enabled bool) { } func (bs *SqliteBlockstore) UpdateRepo(ctx context.Context, root cid.Cid, rev string) error { - if err := bs.db.Exec("UPDATE repos SET root = ?, rev = ? WHERE did = ?", root.Bytes(), rev, bs.did).Error; err != nil { + if err := bs.db.Exec("UPDATE repos SET root = ?, rev = ? WHERE did = ?", nil, root.Bytes(), rev, bs.did).Error; err != nil { return err } diff --git a/cmd/cocoon/main.go b/cmd/cocoon/main.go index 96ce226..9c1a831 100644 --- a/cmd/cocoon/main.go +++ b/cmd/cocoon/main.go @@ -91,6 +91,30 @@ func main() { Required: false, EnvVars: []string{"COCOON_SMTP_NAME"}, }, + &cli.BoolFlag{ + Name: "s3-backups-enabled", + EnvVars: []string{"COCOON_S3_BACKUPS_ENABLED"}, + }, + &cli.StringFlag{ + Name: "s3-region", + EnvVars: []string{"COCOON_S3_REGION"}, + }, + &cli.StringFlag{ + Name: "s3-bucket", + EnvVars: []string{"COCOON_S3_BUCKET"}, + }, + &cli.StringFlag{ + Name: "s3-endpoint", + EnvVars: []string{"COCOON_S3_ENDPOINT"}, + }, + &cli.StringFlag{ + Name: "s3-access-key", + EnvVars: []string{"COCOON_S3_ACCESS_KEY"}, + }, + &cli.StringFlag{ + Name: "s3-secret-key", + EnvVars: []string{"COCOON_S3_SECRET_KEY"}, + }, }, Commands: []*cli.Command{ run, @@ -126,6 +150,14 @@ var run = &cli.Command{ SmtpPort: cmd.String("smtp-port"), SmtpEmail: cmd.String("smtp-email"), SmtpName: cmd.String("smtp-name"), + S3Config: &server.S3Config{ + BackupsEnabled: cmd.Bool("s3-backups-enabled"), + Region: cmd.String("s3-region"), + Bucket: cmd.String("s3-bucket"), + Endpoint: cmd.String("s3-endpoint"), + AccessKey: cmd.String("s3-access-key"), + SecretKey: cmd.String("s3-secret-key"), + }, }) if err != nil { fmt.Printf("error creating cocoon: %v", err) diff --git a/go.mod b/go.mod index 6ea68c0..6004335 100644 --- a/go.mod +++ b/go.mod @@ -4,11 +4,15 @@ go 1.24.1 require ( github.com/Azure/go-autorest/autorest/to v0.4.1 + github.com/aws/aws-sdk-go v1.55.7 github.com/bluesky-social/indigo v0.0.0-20250414202759-826fcdeaa36b github.com/btcsuite/websocket v0.0.0-20150119174127-31079b680792 + github.com/domodwyer/mailyak/v3 v3.6.2 github.com/go-playground/validator v9.31.0+incompatible github.com/golang-jwt/jwt/v4 v4.5.2 github.com/google/uuid v1.4.0 + github.com/gorilla/websocket v1.5.1 + github.com/hashicorp/golang-lru/v2 v2.0.7 github.com/ipfs/go-block-format v0.2.0 github.com/ipfs/go-cid v0.4.1 github.com/ipfs/go-ipld-cbor v0.1.0 @@ -16,8 +20,11 @@ require ( github.com/joho/godotenv v1.5.1 github.com/labstack/echo/v4 v4.13.3 github.com/lestrrat-go/jwx/v2 v2.0.12 + github.com/multiformats/go-multihash v0.2.3 github.com/samber/slog-echo v1.16.1 github.com/urfave/cli/v2 v2.27.6 + github.com/whyrusleeping/cbor-gen v0.2.1-0.20241030202151-b7a6831be65e + gitlab.com/yawning/secp256k1-voi v0.0.0-20230925100816-f2616030848b golang.org/x/crypto v0.36.0 gorm.io/driver/sqlite v1.5.7 gorm.io/gorm v1.25.12 @@ -27,12 +34,10 @@ require ( github.com/Azure/go-autorest v14.2.0+incompatible // indirect github.com/RussellLuo/slidingwindow v0.0.0-20200528002341-535bb99d338b // indirect github.com/beorn7/perks v1.0.1 // indirect - github.com/bradfitz/gomemcache v0.0.0-20230905024940-24af94b03874 // indirect github.com/carlmjohnson/versioninfo v0.22.5 // indirect github.com/cespare/xxhash/v2 v2.2.0 // indirect github.com/cpuguy83/go-md2man/v2 v2.0.5 // indirect github.com/decred/dcrd/dcrec/secp256k1/v4 v4.2.0 // indirect - github.com/domodwyer/mailyak/v3 v3.6.2 // indirect github.com/felixge/httpsnoop v1.0.4 // indirect github.com/go-logr/logr v1.4.2 // indirect github.com/go-logr/stdr v1.2.2 // indirect @@ -41,15 +46,11 @@ require ( github.com/goccy/go-json v0.10.2 // indirect github.com/gocql/gocql v1.7.0 // indirect github.com/gogo/protobuf v1.3.2 // indirect - github.com/golang-jwt/jwt v3.2.2+incompatible // indirect github.com/golang/snappy v0.0.4 // indirect - github.com/gorilla/websocket v1.5.1 // indirect github.com/hailocab/go-hostpool v0.0.0-20160125115350-e80d13ce29ed // indirect github.com/hashicorp/go-cleanhttp v0.5.2 // indirect github.com/hashicorp/go-retryablehttp v0.7.5 // indirect github.com/hashicorp/golang-lru v1.0.2 // indirect - github.com/hashicorp/golang-lru/arc/v2 v2.0.6 // indirect - github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect github.com/ipfs/bbloom v0.0.4 // indirect github.com/ipfs/go-blockservice v0.5.2 // indirect github.com/ipfs/go-datastore v0.6.0 // indirect @@ -65,7 +66,6 @@ require ( github.com/ipfs/go-merkledag v0.11.0 // indirect github.com/ipfs/go-metrics-interface v0.0.1 // indirect github.com/ipfs/go-verifcid v0.0.3 // indirect - github.com/ipld/go-car/v2 v2.13.1 // indirect github.com/ipld/go-codec-dagpb v1.6.0 // indirect github.com/ipld/go-ipld-prime v0.21.0 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect @@ -75,6 +75,7 @@ require ( github.com/jbenet/goprocess v0.1.4 // indirect github.com/jinzhu/inflection v1.0.0 // indirect github.com/jinzhu/now v1.1.5 // indirect + github.com/jmespath/go-jmespath v0.4.0 // indirect github.com/klauspost/cpuid/v2 v2.2.7 // indirect github.com/labstack/gommon v0.4.2 // indirect github.com/leodido/go-urn v1.4.0 // indirect @@ -92,11 +93,8 @@ require ( github.com/multiformats/go-base32 v0.1.0 // indirect github.com/multiformats/go-base36 v0.2.0 // indirect github.com/multiformats/go-multibase v0.2.0 // indirect - github.com/multiformats/go-multicodec v0.9.0 // indirect - github.com/multiformats/go-multihash v0.2.3 // indirect github.com/multiformats/go-varint v0.0.7 // indirect github.com/opentracing/opentracing-go v1.2.0 // indirect - github.com/petar/GoLLRB v0.0.0-20210522233825-ae3b015fd3e9 // indirect github.com/polydawn/refmt v0.89.1-0.20221221234430-40501e09de1f // indirect github.com/prometheus/client_golang v1.17.0 // indirect github.com/prometheus/client_model v0.5.0 // indirect @@ -108,11 +106,7 @@ require ( github.com/spaolacci/murmur3 v1.1.0 // indirect github.com/valyala/bytebufferpool v1.0.0 // indirect github.com/valyala/fasttemplate v1.2.2 // indirect - github.com/whyrusleeping/cbor v0.0.0-20171005072247-63513f603b11 // indirect - github.com/whyrusleeping/cbor-gen v0.2.1-0.20241030202151-b7a6831be65e // indirect - github.com/whyrusleeping/go-did v0.0.0-20230824162731-404d1707d5d6 // indirect github.com/xrash/smetrics v0.0.0-20240521201337-686a1a2994c1 // indirect - gitlab.com/yawning/secp256k1-voi v0.0.0-20230925100816-f2616030848b // indirect gitlab.com/yawning/tuplehash v0.0.0-20230713102510-df83abbf9a02 // indirect go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.46.1 // indirect go.opentelemetry.io/otel v1.29.0 // indirect @@ -121,7 +115,6 @@ require ( go.uber.org/atomic v1.11.0 // indirect go.uber.org/multierr v1.11.0 // indirect go.uber.org/zap v1.26.0 // indirect - golang.org/x/exp v0.0.0-20231110203233-9a3e6036ecaa // indirect golang.org/x/net v0.33.0 // indirect golang.org/x/sync v0.12.0 // indirect golang.org/x/sys v0.31.0 // indirect diff --git a/go.sum b/go.sum index be6c4f8..0fafbdb 100644 --- a/go.sum +++ b/go.sum @@ -7,6 +7,8 @@ github.com/RussellLuo/slidingwindow v0.0.0-20200528002341-535bb99d338b h1:5/++qT github.com/RussellLuo/slidingwindow v0.0.0-20200528002341-535bb99d338b/go.mod h1:4+EPqMRApwwE/6yo6CxiHoSnBzjRr3jsqer7frxP8y4= github.com/alexbrainman/goissue34681 v0.0.0-20191006012335-3fc7a47baff5 h1:iW0a5ljuFxkLGPNem5Ui+KBjFJzKg4Fv2fnxe4dvzpM= github.com/alexbrainman/goissue34681 v0.0.0-20191006012335-3fc7a47baff5/go.mod h1:Y2QMoi1vgtOIfc+6DhrMOGkLoGzqSV2rKp4Sm+opsyA= +github.com/aws/aws-sdk-go v1.55.7 h1:UJrkFq7es5CShfBwlWAC8DA077vp8PyVbQd3lqLiztE= +github.com/aws/aws-sdk-go v1.55.7/go.mod h1:eRwEWoyTWFMVYVQzKMNHWP5/RV4xIUGMQfXQHfHkpNU= github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA= github.com/benbjohnson/clock v1.3.0 h1:ip6w0uFQkncKQ979AypyG0ER7mqUSBdKLOgAle/AT8A= github.com/benbjohnson/clock v1.3.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA= @@ -14,14 +16,10 @@ github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/bitly/go-hostpool v0.0.0-20171023180738-a3a6125de932 h1:mXoPYz/Ul5HYEDvkta6I8/rnYM5gSdSV2tJ6XbZuEtY= github.com/bitly/go-hostpool v0.0.0-20171023180738-a3a6125de932/go.mod h1:NOuUCSz6Q9T7+igc/hlvDOUdtWKryOrtFyIVABv/p7k= -github.com/bluesky-social/indigo v0.0.0-20250322011324-8e3fa7af986a h1:clnSZRgkiifbvfqu9++OHfIh2DWuIoZ8CucxLueQxO0= -github.com/bluesky-social/indigo v0.0.0-20250322011324-8e3fa7af986a/go.mod h1:NVBwZvbBSa93kfyweAmKwOLYawdVHdwZ9s+GZtBBVLA= github.com/bluesky-social/indigo v0.0.0-20250414202759-826fcdeaa36b h1:elwfbe+W7GkUmPKFX1h7HaeHvC/kC0XJWfiEHC62xPg= github.com/bluesky-social/indigo v0.0.0-20250414202759-826fcdeaa36b/go.mod h1:yjdhLA1LkK8VDS/WPUoYPo25/Hq/8rX38Ftr67EsqKY= github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 h1:DDGfHa7BWjL4YnC6+E63dPcxHo2sUxDIu8g3QgEJdRY= github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869/go.mod h1:Ekp36dRnpXw/yCqJaO+ZrUyxD+3VXMFFr56k5XYrpB4= -github.com/bradfitz/gomemcache v0.0.0-20230905024940-24af94b03874 h1:N7oVaKyGp8bttX0bfZGmcGkjz7DLQXhAn3DNd3T0ous= -github.com/bradfitz/gomemcache v0.0.0-20230905024940-24af94b03874/go.mod h1:r5xuitiExdLAJ09PR7vBVENGvp4ZuTBeWTGtxuX3K+c= github.com/btcsuite/websocket v0.0.0-20150119174127-31079b680792 h1:R8vQdOQdZ9Y3SkEwmHoWBmX1DNXhXZqlTpq6s4tyJGc= github.com/btcsuite/websocket v0.0.0-20150119174127-31079b680792/go.mod h1:ghJtEyQwv5/p4Mg4C0fgbePVuGr935/5ddU9Z3TmDRY= github.com/carlmjohnson/versioninfo v0.22.5 h1:O00sjOLUAFxYQjlN/bzYTuZiS0y6fWDQjMRvwtKgwwc= @@ -65,8 +63,6 @@ github.com/gocql/gocql v1.7.0 h1:O+7U7/1gSN7QTEAaMEsJc1Oq2QHXvCWoF3DFK9HDHus= github.com/gocql/gocql v1.7.0/go.mod h1:vnlvXyFZeLBF0Wy+RS8hrOdbn0UWsWtdg07XJnFxZ+4= github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= -github.com/golang-jwt/jwt v3.2.2+incompatible h1:IfV12K8xAKAnZqdXVzCZ+TOjboZ2keLg81eXfW3O+oY= -github.com/golang-jwt/jwt v3.2.2+incompatible/go.mod h1:8pz2t5EyA70fFQQSrl6XZXzqecmYZeUEB8OUGHkxJ+I= github.com/golang-jwt/jwt/v4 v4.5.2 h1:YtQM7lnr8iZ+j5q71MGKkNw9Mn7AjHM68uc9g5fXeUI= github.com/golang-jwt/jwt/v4 v4.5.2/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w39/MY0Ch0= github.com/golang/snappy v0.0.3/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= @@ -93,16 +89,12 @@ github.com/hashicorp/go-retryablehttp v0.7.5 h1:bJj+Pj19UZMIweq/iie+1u5YCdGrnxCT github.com/hashicorp/go-retryablehttp v0.7.5/go.mod h1:Jy/gPYAdjqffZ/yFGCFV2doI5wjtH1ewM9u8iYVjtX8= github.com/hashicorp/golang-lru v1.0.2 h1:dV3g9Z/unq5DpblPpw+Oqcv4dU/1omnb4Ok8iPY6p1c= github.com/hashicorp/golang-lru v1.0.2/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4= -github.com/hashicorp/golang-lru/arc/v2 v2.0.6 h1:4NU7uP5vSoK6TbaMj3NtY478TTAWLso/vL1gpNrInHg= -github.com/hashicorp/golang-lru/arc/v2 v2.0.6/go.mod h1:cfdDIX05DWvYV6/shsxDfa/OVcRieOt+q4FnM8x+Xno= github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= github.com/huin/goupnp v1.0.3 h1:N8No57ls+MnjlB+JPiCVSOyy/ot7MJTqlo7rn+NYSqQ= github.com/huin/goupnp v1.0.3/go.mod h1:ZxNlw5WqJj6wSsRK5+YfflQGXYfccj5VgQsMNixHM7Y= github.com/ipfs/bbloom v0.0.4 h1:Gi+8EGJ2y5qiD5FbsbpX/TMNcJw8gSqr7eyjHa4Fhvs= github.com/ipfs/bbloom v0.0.4/go.mod h1:cS9YprKXpoZ9lT0n/Mw/a6/aFV6DTjTLYHeA+gyqMG0= -github.com/ipfs/go-bitfield v1.1.0 h1:fh7FIo8bSwaJEh6DdTWbCeZ1eqOaOkKFI74SCnsWbGA= -github.com/ipfs/go-bitfield v1.1.0/go.mod h1:paqf1wjq/D2BBmzfTVFlJQ9IlFOZpg422HL0HqsGWHU= github.com/ipfs/go-bitswap v0.11.0 h1:j1WVvhDX1yhG32NTC9xfxnqycqYIlhzEzLXG/cU1HyQ= github.com/ipfs/go-bitswap v0.11.0/go.mod h1:05aE8H3XOU+LXpTedeAS0OZpcO1WFsj5niYQH9a1Tmk= github.com/ipfs/go-block-format v0.2.0 h1:ZqrkxBA2ICbDRbK8KJs/u0O3dlp6gmAuuXUJNiW1Ycs= @@ -123,8 +115,6 @@ github.com/ipfs/go-ipfs-blockstore v1.3.1 h1:cEI9ci7V0sRNivqaOr0elDsamxXFxJMMMy7 github.com/ipfs/go-ipfs-blockstore v1.3.1/go.mod h1:KgtZyc9fq+P2xJUiCAzbRdhhqJHvsw8u2Dlqy2MyRTE= github.com/ipfs/go-ipfs-blocksutil v0.0.1 h1:Eh/H4pc1hsvhzsQoMEP3Bke/aW5P5rVM1IWFJMcGIPQ= github.com/ipfs/go-ipfs-blocksutil v0.0.1/go.mod h1:Yq4M86uIOmxmGPUHv/uI7uKqZNtLb449gwKqXjIsnRk= -github.com/ipfs/go-ipfs-chunker v0.0.5 h1:ojCf7HV/m+uS2vhUGWcogIIxiO5ubl5O57Q7NapWLY8= -github.com/ipfs/go-ipfs-chunker v0.0.5/go.mod h1:jhgdF8vxRHycr00k13FM8Y0E+6BoalYeobXmUyTreP8= github.com/ipfs/go-ipfs-delay v0.0.1 h1:r/UXYyRcddO6thwOnhiznIAiSvxMECGgtv35Xs1IeRQ= github.com/ipfs/go-ipfs-delay v0.0.1/go.mod h1:8SP1YXK1M1kXuc4KJZINY3TQQ03J2rwBG9QfXmbRPrw= github.com/ipfs/go-ipfs-ds-help v1.1.1 h1:B5UJOH52IbcfS56+Ul+sv8jnIV10lbjLF5eOO0C66Nw= @@ -158,8 +148,6 @@ github.com/ipfs/go-metrics-interface v0.0.1 h1:j+cpbjYvu4R8zbleSs36gvB7jR+wsL2fG github.com/ipfs/go-metrics-interface v0.0.1/go.mod h1:6s6euYU4zowdslK0GKHmqaIZ3j/b/tL7HTWtJ4VPgWY= github.com/ipfs/go-peertaskqueue v0.8.1 h1:YhxAs1+wxb5jk7RvS0LHdyiILpNmRIRnZVztekOF0pg= github.com/ipfs/go-peertaskqueue v0.8.1/go.mod h1:Oxxd3eaK279FxeydSPPVGHzbwVeHjatZ2GA8XD+KbPU= -github.com/ipfs/go-unixfsnode v1.8.0 h1:yCkakzuE365glu+YkgzZt6p38CSVEBPgngL9ZkfnyQU= -github.com/ipfs/go-unixfsnode v1.8.0/go.mod h1:HxRu9HYHOjK6HUqFBAi++7DVoWAHn0o4v/nZ/VA+0g8= github.com/ipfs/go-verifcid v0.0.3 h1:gmRKccqhWDocCRkC+a59g5QW7uJw5bpX9HWBevXa0zs= github.com/ipfs/go-verifcid v0.0.3/go.mod h1:gcCtGniVzelKrbk9ooUSX/pM3xlH73fZZJDzQJRvOUw= github.com/ipld/go-car v0.6.1-0.20230509095817-92d28eb23ba4 h1:oFo19cBmcP0Cmg3XXbrr0V/c+xU9U1huEZp8+OgBzdI= @@ -170,8 +158,6 @@ github.com/ipld/go-codec-dagpb v1.6.0 h1:9nYazfyu9B1p3NAgfVdpRco3Fs2nFC72DqVsMj6 github.com/ipld/go-codec-dagpb v1.6.0/go.mod h1:ANzFhfP2uMJxRBr8CE+WQWs5UsNa0pYtmKZ+agnUw9s= github.com/ipld/go-ipld-prime v0.21.0 h1:n4JmcpOlPDIxBcY037SVfpd1G+Sj1nKZah0m6QH9C2E= github.com/ipld/go-ipld-prime v0.21.0/go.mod h1:3RLqy//ERg/y5oShXXdx5YIp50cFGOanyMctpPjsvxQ= -github.com/ipld/go-ipld-prime/storage/bsadapter v0.0.0-20230102063945-1a409dc236dd h1:gMlw/MhNr2Wtp5RwGdsW23cs+yCuj9k2ON7i9MiJlRo= -github.com/ipld/go-ipld-prime/storage/bsadapter v0.0.0-20230102063945-1a409dc236dd/go.mod h1:wZ8hH8UxeryOs4kJEJaiui/s00hDSbE37OKsL47g+Sw= github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk= @@ -189,6 +175,10 @@ github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc= github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ= github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8= +github.com/jmespath/go-jmespath v0.4.0 h1:BEgLn5cpjn8UN1mAw4NjwDrS35OdebyEtFe+9YPoQUg= +github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo= +github.com/jmespath/go-jmespath/internal/testify v1.5.1 h1:shLQSRRSCCPj3f2gpwzGwWFoC7ycTf1rcQZHOlsJ6N8= +github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U= github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0= github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4= github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo= @@ -281,8 +271,6 @@ github.com/multiformats/go-varint v0.0.7 h1:sWSGR+f/eu5ABZA2ZpYKBILXTTs9JWpdEM/n github.com/multiformats/go-varint v0.0.7/go.mod h1:r8PUYw/fD/SjBCiKOoDlGF6QawOELpZAu9eioSos/OU= github.com/opentracing/opentracing-go v1.2.0 h1:uEJPy/1a5RIPAJ0Ov+OIO8OxWu77jEv+1B0VhjKrZUs= github.com/opentracing/opentracing-go v1.2.0/go.mod h1:GxEUsuufX4nBwe+T+Wl9TAgYrxe9dPLANfrWvHYVTgc= -github.com/orandin/slog-gorm v1.3.2 h1:C0lKDQPAx/pF+8K2HL7bdShPwOEJpPM0Bn80zTzxU1g= -github.com/orandin/slog-gorm v1.3.2/go.mod h1:MoZ51+b7xE9lwGNPYEhxcUtRNrYzjdcKvA8QXQQGEPA= github.com/petar/GoLLRB v0.0.0-20210522233825-ae3b015fd3e9 h1:1/WtZae0yGtPq+TI6+Tv1WTxkukpXeMlviSxvL7SRgk= github.com/petar/GoLLRB v0.0.0-20210522233825-ae3b015fd3e9/go.mod h1:x3N5drFsm2uilKKuuYo6LdyD8vZAW55sH/9w+pbo1sw= github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= @@ -345,10 +333,6 @@ github.com/whyrusleeping/cbor v0.0.0-20171005072247-63513f603b11 h1:5HZfQkwe0mIf github.com/whyrusleeping/cbor v0.0.0-20171005072247-63513f603b11/go.mod h1:Wlo/SzPmxVp6vXpGt/zaXhHH0fn4IxgqZc82aKg6bpQ= github.com/whyrusleeping/cbor-gen v0.2.1-0.20241030202151-b7a6831be65e h1:28X54ciEwwUxyHn9yrZfl5ojgF4CBNLWX7LR0rvBkf4= github.com/whyrusleeping/cbor-gen v0.2.1-0.20241030202151-b7a6831be65e/go.mod h1:pM99HXyEbSQHcosHc0iW7YFmwnscr+t9Te4ibko05so= -github.com/whyrusleeping/chunker v0.0.0-20181014151217-fe64bd25879f h1:jQa4QT2UP9WYv2nzyawpKMOCl+Z/jW7djv2/J50lj9E= -github.com/whyrusleeping/chunker v0.0.0-20181014151217-fe64bd25879f/go.mod h1:p9UJB6dDgdPgMJZs7UjUOdulKyRr9fqkS+6JKAInPy8= -github.com/whyrusleeping/go-did v0.0.0-20230824162731-404d1707d5d6 h1:yJ9/LwIGIk/c0CdoavpC9RNSGSruIspSZtxG3Nnldic= -github.com/whyrusleeping/go-did v0.0.0-20230824162731-404d1707d5d6/go.mod h1:39U9RRVr4CKbXpXYopWn+FSH5s+vWu6+RmguSPWAq5s= github.com/xrash/smetrics v0.0.0-20240521201337-686a1a2994c1 h1:gEOO8jv9F4OT7lGCjxCBTO/36wtF6j2nSip77qHd4x4= github.com/xrash/smetrics v0.0.0-20240521201337-686a1a2994c1/go.mod h1:Ohn+xnUBiLI6FVj/9LpzZWtj1/D6lUovWYBkxHVV3aM= github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= diff --git a/internal/db/db.go b/internal/db/db.go new file mode 100644 index 0000000..709c600 --- /dev/null +++ b/internal/db/db.go @@ -0,0 +1,65 @@ +package db + +import ( + "sync" + + "gorm.io/gorm" + "gorm.io/gorm/clause" +) + +type DB struct { + cli *gorm.DB + mu sync.Mutex +} + +func NewDB(cli *gorm.DB) *DB { + return &DB{ + cli: cli, + mu: sync.Mutex{}, + } +} + +func (db *DB) Create(value any, clauses []clause.Expression) *gorm.DB { + db.mu.Lock() + defer db.mu.Unlock() + return db.cli.Clauses(clauses...).Create(value) +} + +func (db *DB) Exec(sql string, clauses []clause.Expression, values ...any) *gorm.DB { + db.mu.Lock() + defer db.mu.Unlock() + return db.cli.Clauses(clauses...).Exec(sql, values...) +} + +func (db *DB) Raw(sql string, clauses []clause.Expression, values ...any) *gorm.DB { + return db.cli.Clauses(clauses...).Raw(sql, values...) +} + +func (db *DB) AutoMigrate(models ...any) error { + return db.cli.AutoMigrate(models...) +} + +func (db *DB) Delete(value any, clauses []clause.Expression) *gorm.DB { + db.mu.Lock() + defer db.mu.Unlock() + return db.cli.Clauses(clauses...).Delete(value) +} + +func (db *DB) First(dest any, conds ...any) *gorm.DB { + return db.cli.First(dest, conds...) +} + +// TODO: this isn't actually good. we can commit even if the db is locked here. this is probably okay for the time being, but need to figure +// out a better solution. right now we only do this whenever we're importing a repo though so i'm mostly not worried, but it's still bad. +// e.g. when we do apply writes we should also be using a transcation but we don't right now +func (db *DB) BeginDangerously() *gorm.DB { + return db.cli.Begin() +} + +func (db *DB) Lock() { + db.mu.Lock() +} + +func (db *DB) Unlock() { + db.mu.Unlock() +} diff --git a/server/common.go b/server/common.go index d79ea46..3905767 100644 --- a/server/common.go +++ b/server/common.go @@ -22,7 +22,7 @@ func (s *Server) getRepoByEmail(email string) (*models.Repo, error) { func (s *Server) getRepoActorByEmail(email string) (*models.RepoActor, error) { var repo models.RepoActor - if err := s.db.Raw("SELECT r.*, a.* FROM repos r LEFT JOIN actors a ON r.did = a.did WHERE r.email= ?", email).Scan(&repo).Error; err != nil { + if err := s.db.Raw("SELECT r.*, a.* FROM repos r LEFT JOIN actors a ON r.did = a.did WHERE r.email= ?", nil, email).Scan(&repo).Error; err != nil { return nil, err } return &repo, nil @@ -30,7 +30,7 @@ func (s *Server) getRepoActorByEmail(email string) (*models.RepoActor, error) { func (s *Server) getRepoActorByDid(did string) (*models.RepoActor, error) { var repo models.RepoActor - if err := s.db.Raw("SELECT r.*, a.* FROM repos r LEFT JOIN actors a ON r.did = a.did WHERE r.did = ?", did).Scan(&repo).Error; err != nil { + if err := s.db.Raw("SELECT r.*, a.* FROM repos r LEFT JOIN actors a ON r.did = a.did WHERE r.did = ?", nil, did).Scan(&repo).Error; err != nil { return nil, err } return &repo, nil diff --git a/server/handle_actor_put_preferences.go b/server/handle_actor_put_preferences.go index 5066ca8..2cb1423 100644 --- a/server/handle_actor_put_preferences.go +++ b/server/handle_actor_put_preferences.go @@ -22,7 +22,7 @@ func (s *Server) handleActorPutPreferences(e echo.Context) error { return err } - if err := s.db.Exec("UPDATE repos SET preferences = ? WHERE did = ?", b, repo.Repo.Did).Error; err != nil { + if err := s.db.Exec("UPDATE repos SET preferences = ? WHERE did = ?", nil, b, repo.Repo.Did).Error; err != nil { return err } diff --git a/server/handle_identity_update_handle.go b/server/handle_identity_update_handle.go index db117b4..ec5376c 100644 --- a/server/handle_identity_update_handle.go +++ b/server/handle_identity_update_handle.go @@ -103,7 +103,7 @@ func (s *Server) handleIdentityUpdateHandle(e echo.Context) error { }, }) - if err := s.db.Exec("UPDATE actors SET handle = ? WHERE did = ?", req.Handle, repo.Repo.Did).Error; err != nil { + if err := s.db.Exec("UPDATE actors SET handle = ? WHERE did = ?", nil, req.Handle, repo.Repo.Did).Error; err != nil { s.logger.Error("error updating handle in db", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_import_repo.go b/server/handle_import_repo.go index 1e66cba..2104c88 100644 --- a/server/handle_import_repo.go +++ b/server/handle_import_repo.go @@ -64,7 +64,7 @@ func (s *Server) handleRepoImportRepo(e echo.Context) error { return helpers.ServerError(e, nil) } - tx := s.db.Begin() + tx := s.db.BeginDangerously() clock := syntax.NewTIDClock(0) diff --git a/server/handle_repo_describe_repo.go b/server/handle_repo_describe_repo.go index ff350f6..ffa530f 100644 --- a/server/handle_repo_describe_repo.go +++ b/server/handle_repo_describe_repo.go @@ -64,7 +64,7 @@ func (s *Server) handleDescribeRepo(e echo.Context) error { } var records []models.Record - if err := s.db.Raw("SELECT DISTINCT(nsid) FROM records WHERE did = ?", repo.Repo.Did).Scan(&records).Error; err != nil { + if err := s.db.Raw("SELECT DISTINCT(nsid) FROM records WHERE did = ?", nil, repo.Repo.Did).Scan(&records).Error; err != nil { s.logger.Error("error getting collections", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_repo_get_record.go b/server/handle_repo_get_record.go index 44f35dc..a99c912 100644 --- a/server/handle_repo_get_record.go +++ b/server/handle_repo_get_record.go @@ -32,7 +32,7 @@ func (s *Server) handleRepoGetRecord(e echo.Context) error { } var record models.Record - if err := s.db.Raw("SELECT * FROM records WHERE did = ? AND nsid = ? AND rkey = ?"+cidquery, params...).Scan(&record).Error; err != nil { + if err := s.db.Raw("SELECT * FROM records WHERE did = ? AND nsid = ? AND rkey = ?"+cidquery, nil, params...).Scan(&record).Error; err != nil { // TODO: handle error nicely return err } diff --git a/server/handle_repo_list_records.go b/server/handle_repo_list_records.go index 94d0444..3c7da6c 100644 --- a/server/handle_repo_list_records.go +++ b/server/handle_repo_list_records.go @@ -64,7 +64,7 @@ func (s *Server) handleListRecords(e echo.Context) error { params = append(params, limit) var records []models.Record - if err := s.db.Raw("SELECT * FROM records WHERE did = ? AND nsid = ? "+cursorquery+" ORDER BY created_at "+sort+" limit ?", params...).Scan(&records).Error; err != nil { + if err := s.db.Raw("SELECT * FROM records WHERE did = ? AND nsid = ? "+cursorquery+" ORDER BY created_at "+sort+" limit ?", nil, params...).Scan(&records).Error; err != nil { s.logger.Error("error getting records", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_repo_list_repos.go b/server/handle_repo_list_repos.go index 2b91fe8..95845b6 100644 --- a/server/handle_repo_list_repos.go +++ b/server/handle_repo_list_repos.go @@ -22,7 +22,7 @@ type ComAtprotoSyncListReposRepoItem struct { // TODO: paginate this bitch func (s *Server) handleListRepos(e echo.Context) error { var repos []models.Repo - if err := s.db.Raw("SELECT * FROM repos ORDER BY created_at DESC LIMIT 500").Scan(&repos).Error; err != nil { + if err := s.db.Raw("SELECT * FROM repos ORDER BY created_at DESC LIMIT 500", nil).Scan(&repos).Error; err != nil { return err } diff --git a/server/handle_repo_upload_blob.go b/server/handle_repo_upload_blob.go index a350d34..f86e91b 100644 --- a/server/handle_repo_upload_blob.go +++ b/server/handle_repo_upload_blob.go @@ -40,7 +40,7 @@ func (s *Server) handleRepoUploadBlob(e echo.Context) error { CreatedAt: s.repoman.clock.Next().String(), } - if err := s.db.Create(&blob).Error; err != nil { + if err := s.db.Create(&blob, nil).Error; err != nil { s.logger.Error("error creating new blob in db", "error", err) return helpers.ServerError(e, nil) } @@ -72,7 +72,7 @@ func (s *Server) handleRepoUploadBlob(e echo.Context) error { Data: data, } - if err := s.db.Create(&blobPart).Error; err != nil { + if err := s.db.Create(&blobPart, nil).Error; err != nil { s.logger.Error("error adding blob part to db", "error", err) return helpers.ServerError(e, nil) } @@ -89,7 +89,7 @@ func (s *Server) handleRepoUploadBlob(e echo.Context) error { return helpers.ServerError(e, nil) } - if err := s.db.Exec("UPDATE blobs SET cid = ? WHERE id = ?", c.Bytes(), blob.ID).Error; err != nil { + if err := s.db.Exec("UPDATE blobs SET cid = ? WHERE id = ?", nil, c.Bytes(), blob.ID).Error; err != nil { // there should probably be somme handling here if this fails... s.logger.Error("error updating blob", "error", err) return helpers.ServerError(e, nil) diff --git a/server/handle_server_confirm_email.go b/server/handle_server_confirm_email.go index 734941c..bf4e05f 100644 --- a/server/handle_server_confirm_email.go +++ b/server/handle_server_confirm_email.go @@ -41,7 +41,7 @@ func (s *Server) handleServerConfirmEmail(e echo.Context) error { now := time.Now().UTC() - if err := s.db.Exec("UPDATE repos SET email_verification_code = NULL, email_verification_code_expires_at = NULL, email_confirmed_at = ? WHERE did = ?", now, urepo.Repo.Did).Error; err != nil { + if err := s.db.Exec("UPDATE repos SET email_verification_code = NULL, email_verification_code_expires_at = NULL, email_confirmed_at = ? WHERE did = ?", nil, now, urepo.Repo.Did).Error; err != nil { s.logger.Error("error updating user", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_server_create_account.go b/server/handle_server_create_account.go index 1afad53..47bcaee 100644 --- a/server/handle_server_create_account.go +++ b/server/handle_server_create_account.go @@ -102,7 +102,7 @@ func (s *Server) handleCreateAccount(e echo.Context) error { } var ic models.InviteCode - if err := s.db.Raw("SELECT * FROM invite_codes WHERE code = ?", request.InviteCode).Scan(&ic).Error; err != nil { + if err := s.db.Raw("SELECT * FROM invite_codes WHERE code = ?", nil, request.InviteCode).Scan(&ic).Error; err != nil { if err == gorm.ErrRecordNotFound { return helpers.InputError(e, to.StringPtr("InvalidInviteCode")) } @@ -166,12 +166,12 @@ func (s *Server) handleCreateAccount(e echo.Context) error { Handle: request.Handle, } - if err := s.db.Create(&urepo).Error; err != nil { + if err := s.db.Create(&urepo, nil).Error; err != nil { s.logger.Error("error inserting new repo", "error", err) return helpers.ServerError(e, nil) } - if err := s.db.Create(&actor).Error; err != nil { + if err := s.db.Create(&actor, nil).Error; err != nil { s.logger.Error("error inserting new actor", "error", err) return helpers.ServerError(e, nil) } @@ -210,7 +210,7 @@ func (s *Server) handleCreateAccount(e echo.Context) error { }) } - if err := s.db.Raw("UPDATE invite_codes SET remaining_use_count = remaining_use_count - 1 WHERE code = ?", request.InviteCode).Scan(&ic).Error; err != nil { + if err := s.db.Raw("UPDATE invite_codes SET remaining_use_count = remaining_use_count - 1 WHERE code = ?", nil, request.InviteCode).Scan(&ic).Error; err != nil { s.logger.Error("error decrementing use count", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_server_create_invite_code.go b/server/handle_server_create_invite_code.go index d179cdf..53c2570 100644 --- a/server/handle_server_create_invite_code.go +++ b/server/handle_server_create_invite_code.go @@ -41,7 +41,7 @@ func (s *Server) handleCreateInviteCode(e echo.Context) error { Code: ic, Did: acc, RemainingUseCount: req.UseCount, - }).Error; err != nil { + }, nil).Error; err != nil { s.logger.Error("error creating invite code", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_server_create_invite_codes.go b/server/handle_server_create_invite_codes.go index 261d841..7d13fc5 100644 --- a/server/handle_server_create_invite_codes.go +++ b/server/handle_server_create_invite_codes.go @@ -54,7 +54,7 @@ func (s *Server) handleCreateInviteCodes(e echo.Context) error { Code: ic, Did: did, RemainingUseCount: req.UseCount, - }).Error; err != nil { + }, nil).Error; err != nil { s.logger.Error("error creating invite code", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_server_create_session.go b/server/handle_server_create_session.go index dcbf6bb..847557f 100644 --- a/server/handle_server_create_session.go +++ b/server/handle_server_create_session.go @@ -65,11 +65,11 @@ func (s *Server) handleCreateSession(e echo.Context) error { var err error switch idtype { case "did": - err = s.db.Raw("SELECT r.*, a.* FROM repos r LEFT JOIN actors a ON r.did = a.did WHERE r.did = ?", req.Identifier).Scan(&repo).Error + err = s.db.Raw("SELECT r.*, a.* FROM repos r LEFT JOIN actors a ON r.did = a.did WHERE r.did = ?", nil, req.Identifier).Scan(&repo).Error case "handle": - err = s.db.Raw("SELECT r.*, a.* FROM actors a LEFT JOIN repos r ON a.did = r.did WHERE a.handle = ?", req.Identifier).Scan(&repo).Error + err = s.db.Raw("SELECT r.*, a.* FROM actors a LEFT JOIN repos r ON a.did = r.did WHERE a.handle = ?", nil, req.Identifier).Scan(&repo).Error case "email": - err = s.db.Raw("SELECT r.*, a.* FROM repos r LEFT JOIN actors a ON r.did = a.did WHERE r.email = ?", req.Identifier).Scan(&repo).Error + err = s.db.Raw("SELECT r.*, a.* FROM repos r LEFT JOIN actors a ON r.did = a.did WHERE r.email = ?", nil, req.Identifier).Scan(&repo).Error } if err != nil { diff --git a/server/handle_server_delete_session.go b/server/handle_server_delete_session.go index 5ceb060..bd054fb 100644 --- a/server/handle_server_delete_session.go +++ b/server/handle_server_delete_session.go @@ -10,12 +10,12 @@ func (s *Server) handleDeleteSession(e echo.Context) error { token := e.Get("token").(string) var acctok models.Token - if err := s.db.Raw("DELETE FROM tokens WHERE token = ? RETURNING *", token).Scan(&acctok).Error; err != nil { + if err := s.db.Raw("DELETE FROM tokens WHERE token = ? RETURNING *", nil, token).Scan(&acctok).Error; err != nil { s.logger.Error("error deleting access token from db", "error", err) return helpers.ServerError(e, nil) } - if err := s.db.Exec("DELETE FROM refresh_tokens WHERE token = ?", acctok.RefreshToken).Error; err != nil { + if err := s.db.Exec("DELETE FROM refresh_tokens WHERE token = ?", nil, acctok.RefreshToken).Error; err != nil { s.logger.Error("error deleting refresh token from db", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_server_refresh_session.go b/server/handle_server_refresh_session.go index d36e96a..5a0a9e1 100644 --- a/server/handle_server_refresh_session.go +++ b/server/handle_server_refresh_session.go @@ -19,12 +19,12 @@ func (s *Server) handleRefreshSession(e echo.Context) error { token := e.Get("token").(string) repo := e.Get("repo").(*models.RepoActor) - if err := s.db.Exec("DELETE FROM refresh_tokens WHERE token = ?", token).Error; err != nil { + if err := s.db.Exec("DELETE FROM refresh_tokens WHERE token = ?", nil, token).Error; err != nil { s.logger.Error("error getting refresh token from db", "error", err) return helpers.ServerError(e, nil) } - if err := s.db.Exec("DELETE FROM tokens WHERE refresh_token = ?", token).Error; err != nil { + if err := s.db.Exec("DELETE FROM tokens WHERE refresh_token = ?", nil, token).Error; err != nil { s.logger.Error("error deleting access token from db", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_server_request_email_confirmation.go b/server/handle_server_request_email_confirmation.go index 8daceeb..d412f86 100644 --- a/server/handle_server_request_email_confirmation.go +++ b/server/handle_server_request_email_confirmation.go @@ -20,7 +20,7 @@ func (s *Server) handleServerRequestEmailConfirmation(e echo.Context) error { code := fmt.Sprintf("%s-%s", helpers.RandomVarchar(5), helpers.RandomVarchar(5)) eat := time.Now().Add(10 * time.Minute).UTC() - if err := s.db.Exec("UPDATE repos SET email_verification_code = ?, email_verification_code_expires_at = ? WHERE did = ?", code, eat, urepo.Repo.Did).Error; err != nil { + if err := s.db.Exec("UPDATE repos SET email_verification_code = ?, email_verification_code_expires_at = ? WHERE did = ?", nil, code, eat, urepo.Repo.Did).Error; err != nil { s.logger.Error("error updating user", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_server_request_email_update.go b/server/handle_server_request_email_update.go index cc7f11c..3af87e1 100644 --- a/server/handle_server_request_email_update.go +++ b/server/handle_server_request_email_update.go @@ -20,7 +20,7 @@ func (s *Server) handleServerRequestEmailUpdate(e echo.Context) error { code := fmt.Sprintf("%s-%s", helpers.RandomVarchar(5), helpers.RandomVarchar(5)) eat := time.Now().Add(10 * time.Minute).UTC() - if err := s.db.Exec("UPDATE repos SET email_update_code = ?, email_update_code_expires_at = ? WHERE did = ?", code, eat, urepo.Repo.Did).Error; err != nil { + if err := s.db.Exec("UPDATE repos SET email_update_code = ?, email_update_code_expires_at = ? WHERE did = ?", nil, code, eat, urepo.Repo.Did).Error; err != nil { s.logger.Error("error updating repo", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_server_request_password_reset.go b/server/handle_server_request_password_reset.go index d54e99f..6339a18 100644 --- a/server/handle_server_request_password_reset.go +++ b/server/handle_server_request_password_reset.go @@ -36,7 +36,7 @@ func (s *Server) handleServerRequestPasswordReset(e echo.Context) error { code := fmt.Sprintf("%s-%s", helpers.RandomVarchar(5), helpers.RandomVarchar(5)) eat := time.Now().Add(10 * time.Minute).UTC() - if err := s.db.Exec("UPDATE repos SET password_reset_code = ?, password_reset_code_expires_at = ? WHERE did = ?", code, eat, urepo.Repo.Did).Error; err != nil { + if err := s.db.Exec("UPDATE repos SET password_reset_code = ?, password_reset_code_expires_at = ? WHERE did = ?", nil, code, eat, urepo.Repo.Did).Error; err != nil { s.logger.Error("error updating repo", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_server_reset_password.go b/server/handle_server_reset_password.go index 1955652..3ba64dc 100644 --- a/server/handle_server_reset_password.go +++ b/server/handle_server_reset_password.go @@ -46,7 +46,7 @@ func (s *Server) handleServerResetPassword(e echo.Context) error { return helpers.ServerError(e, nil) } - if err := s.db.Exec("UPDATE repos SET password_reset_code = NULL, password_reset_code_expires_at = NULL, password = ? WHERE did = ?", hash, urepo.Repo.Did).Error; err != nil { + if err := s.db.Exec("UPDATE repos SET password_reset_code = NULL, password_reset_code_expires_at = NULL, password = ? WHERE did = ?", nil, hash, urepo.Repo.Did).Error; err != nil { s.logger.Error("error updating repo", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_server_update_email.go b/server/handle_server_update_email.go index b7203e5..f0bce18 100644 --- a/server/handle_server_update_email.go +++ b/server/handle_server_update_email.go @@ -40,7 +40,7 @@ func (s *Server) handleServerUpdateEmail(e echo.Context) error { return helpers.InputError(e, to.StringPtr("ExpiredToken")) } - if err := s.db.Exec("UPDATE repos SET email_update_code = NULL, email_update_code_expires_at = NULL, email_confirmed_at = NULL, email = ? WHERE did = ?", req.Email, urepo.Repo.Did).Error; err != nil { + if err := s.db.Exec("UPDATE repos SET email_update_code = NULL, email_update_code_expires_at = NULL, email_confirmed_at = NULL, email = ? WHERE did = ?", nil, req.Email, urepo.Repo.Did).Error; err != nil { s.logger.Error("error updating repo", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_sync_get_blob.go b/server/handle_sync_get_blob.go index 0e54c1c..2c1c412 100644 --- a/server/handle_sync_get_blob.go +++ b/server/handle_sync_get_blob.go @@ -26,7 +26,7 @@ func (s *Server) handleSyncGetBlob(e echo.Context) error { } var blob models.Blob - if err := s.db.Raw("SELECT * FROM blobs WHERE did = ? AND cid = ?", did, c.Bytes()).Scan(&blob).Error; err != nil { + if err := s.db.Raw("SELECT * FROM blobs WHERE did = ? AND cid = ?", nil, did, c.Bytes()).Scan(&blob).Error; err != nil { s.logger.Error("error looking up blob", "error", err) return helpers.ServerError(e, nil) } @@ -34,7 +34,7 @@ func (s *Server) handleSyncGetBlob(e echo.Context) error { buf := new(bytes.Buffer) var parts []models.BlobPart - if err := s.db.Raw("SELECT * FROM blob_parts WHERE blob_id = ? ORDER BY idx", blob.ID).Scan(&parts).Error; err != nil { + if err := s.db.Raw("SELECT * FROM blob_parts WHERE blob_id = ? ORDER BY idx", nil, blob.ID).Scan(&parts).Error; err != nil { s.logger.Error("error getting blob parts", "error", err) return helpers.ServerError(e, nil) } @@ -44,7 +44,7 @@ func (s *Server) handleSyncGetBlob(e echo.Context) error { buf.Write(p.Data) } - e.Response().Header().Set(echo.HeaderContentDisposition, "attachment; filename=" + c.String()) + e.Response().Header().Set(echo.HeaderContentDisposition, "attachment; filename="+c.String()) return e.Stream(200, "application/octet-stream", buf) } diff --git a/server/handle_sync_get_record.go b/server/handle_sync_get_record.go index cf9405b..13621ad 100644 --- a/server/handle_sync_get_record.go +++ b/server/handle_sync_get_record.go @@ -18,7 +18,7 @@ func (s *Server) handleSyncGetRecord(e echo.Context) error { rkey := e.QueryParam("rkey") var urepo models.Repo - if err := s.db.Raw("SELECT * FROM repos WHERE did = ?", did).Scan(&urepo).Error; err != nil { + if err := s.db.Raw("SELECT * FROM repos WHERE did = ?", nil, did).Scan(&urepo).Error; err != nil { s.logger.Error("error getting repo", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_sync_get_repo.go b/server/handle_sync_get_repo.go index da511cb..f862233 100644 --- a/server/handle_sync_get_repo.go +++ b/server/handle_sync_get_repo.go @@ -41,7 +41,7 @@ func (s *Server) handleSyncGetRepo(e echo.Context) error { } var blocks []models.Block - if err := s.db.Raw("SELECT * FROM blocks WHERE did = ? ORDER BY rev ASC", urepo.Repo.Did).Scan(&blocks).Error; err != nil { + if err := s.db.Raw("SELECT * FROM blocks WHERE did = ? ORDER BY rev ASC", nil, urepo.Repo.Did).Scan(&blocks).Error; err != nil { return err } diff --git a/server/handle_sync_list_blobs.go b/server/handle_sync_list_blobs.go index d28db73..c867d7a 100644 --- a/server/handle_sync_list_blobs.go +++ b/server/handle_sync_list_blobs.go @@ -35,7 +35,7 @@ func (s *Server) handleSyncListBlobs(e echo.Context) error { params = append(params, limit) var blobs []models.Blob - if err := s.db.Raw("SELECT * FROM blobs WHERE did = ? "+cursorquery+" ORDER BY created_at DESC LIMIT ?", params...).Scan(&blobs).Error; err != nil { + if err := s.db.Raw("SELECT * FROM blobs WHERE did = ? "+cursorquery+" ORDER BY created_at DESC LIMIT ?", nil, params...).Scan(&blobs).Error; err != nil { s.logger.Error("error getting records", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/repo.go b/server/repo.go index bbde68b..1f84e21 100644 --- a/server/repo.go +++ b/server/repo.go @@ -18,17 +18,17 @@ import ( "github.com/bluesky-social/indigo/repo" "github.com/bluesky-social/indigo/util" "github.com/haileyok/cocoon/blockstore" + "github.com/haileyok/cocoon/internal/db" "github.com/haileyok/cocoon/models" blocks "github.com/ipfs/go-block-format" "github.com/ipfs/go-cid" cbor "github.com/ipfs/go-ipld-cbor" "github.com/ipld/go-car" - "gorm.io/gorm" "gorm.io/gorm/clause" ) type RepoMan struct { - db *gorm.DB + db *db.DB s *Server clock *syntax.TIDClock } @@ -162,7 +162,7 @@ func (rm *RepoMan) applyWrites(urepo models.Repo, writes []Op, swapCommit *strin }) case OpTypeDelete: var old models.Record - if err := rm.db.Raw("SELECT value FROM records WHERE did = ? AND nsid = ? AND rkey = ?", urepo.Did, op.Collection, op.Rkey).Scan(&old).Error; err != nil { + if err := rm.db.Raw("SELECT value FROM records WHERE did = ? AND nsid = ? AND rkey = ?", nil, urepo.Did, op.Collection, op.Rkey).Scan(&old).Error; err != nil { return nil, err } entries = append(entries, models.Record{ @@ -284,10 +284,10 @@ func (rm *RepoMan) applyWrites(urepo models.Repo, writes []Op, swapCommit *strin for _, entry := range entries { var cids []cid.Cid if entry.Cid != "" { - if err := rm.s.db.Clauses(clause.OnConflict{ + if err := rm.s.db.Create(&entry, []clause.Expression{clause.OnConflict{ Columns: []clause.Column{{Name: "did"}, {Name: "nsid"}, {Name: "rkey"}}, UpdateAll: true, - }).Create(&entry).Error; err != nil { + }}).Error; err != nil { return nil, err } @@ -296,7 +296,7 @@ func (rm *RepoMan) applyWrites(urepo models.Repo, writes []Op, swapCommit *strin return nil, err } } else { - if err := rm.s.db.Delete(&entry).Error; err != nil { + if err := rm.s.db.Delete(&entry, nil).Error; err != nil { return nil, err } cids, err = rm.decrementBlobRefs(urepo, entry.Value) @@ -368,7 +368,7 @@ func (rm *RepoMan) incrementBlobRefs(urepo models.Repo, cbor []byte) ([]cid.Cid, } for _, c := range cids { - if err := rm.db.Exec("UPDATE blobs SET ref_count = ref_count + 1 WHERE did = ? AND cid = ?", urepo.Did, c.Bytes()).Error; err != nil { + if err := rm.db.Exec("UPDATE blobs SET ref_count = ref_count + 1 WHERE did = ? AND cid = ?", nil, urepo.Did, c.Bytes()).Error; err != nil { return nil, err } } @@ -387,15 +387,15 @@ func (rm *RepoMan) decrementBlobRefs(urepo models.Repo, cbor []byte) ([]cid.Cid, ID uint Count int } - if err := rm.db.Raw("UPDATE blobs SET ref_count = ref_count - 1 WHERE did = ? AND cid = ? RETURNING id, ref_count", urepo.Did, c.Bytes()).Scan(&res).Error; err != nil { + if err := rm.db.Raw("UPDATE blobs SET ref_count = ref_count - 1 WHERE did = ? AND cid = ? RETURNING id, ref_count", nil, urepo.Did, c.Bytes()).Scan(&res).Error; err != nil { return nil, err } if res.Count == 0 { - if err := rm.db.Exec("DELETE FROM blobs WHERE id = ?", res.ID).Error; err != nil { + if err := rm.db.Exec("DELETE FROM blobs WHERE id = ?", nil, res.ID).Error; err != nil { return nil, err } - if err := rm.db.Exec("DELETE FROM blob_parts WHERE blob_id = ?", res.ID).Error; err != nil { + if err := rm.db.Exec("DELETE FROM blob_parts WHERE blob_id = ?", nil, res.ID).Error; err != nil { return nil, err } } diff --git a/server/server.go b/server/server.go index 669227d..f8aa525 100644 --- a/server/server.go +++ b/server/server.go @@ -1,10 +1,12 @@ package server import ( + "bytes" "context" "crypto/ecdsa" "errors" "fmt" + "io" "log/slog" "net/http" "net/smtp" @@ -14,6 +16,10 @@ import ( "time" "github.com/Azure/go-autorest/autorest/to" + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/aws/credentials" + "github.com/aws/aws-sdk-go/aws/session" + "github.com/aws/aws-sdk-go/service/s3" "github.com/bluesky-social/indigo/api/atproto" "github.com/bluesky-social/indigo/atproto/syntax" "github.com/bluesky-social/indigo/events" @@ -23,6 +29,7 @@ import ( "github.com/go-playground/validator" "github.com/golang-jwt/jwt/v4" "github.com/haileyok/cocoon/identity" + "github.com/haileyok/cocoon/internal/db" "github.com/haileyok/cocoon/internal/helpers" "github.com/haileyok/cocoon/models" "github.com/haileyok/cocoon/plc" @@ -34,13 +41,22 @@ import ( "gorm.io/gorm" ) +type S3Config struct { + BackupsEnabled bool + Endpoint string + Region string + Bucket string + AccessKey string + SecretKey string +} + type Server struct { http *http.Client httpd *http.Server mail *mailyak.MailYak mailLk *sync.Mutex echo *echo.Echo - db *gorm.DB + db *db.DB plcClient *plc.Client logger *slog.Logger config *config @@ -48,6 +64,9 @@ type Server struct { repoman *RepoMan evtman *events.EventManager passport *identity.Passport + + dbName string + s3Config *S3Config } type Args struct { @@ -69,6 +88,8 @@ type Args struct { SmtpPort string SmtpEmail string SmtpName string + + S3Config *S3Config } type config struct { @@ -176,7 +197,7 @@ func (s *Server) handleSessionMiddleware(next echo.HandlerFunc) echo.HandlerFunc Found bool } var result Result - if err := s.db.Raw("SELECT EXISTS(SELECT 1 FROM "+table+" WHERE token = ?) AS found", tokenstr).Scan(&result).Error; err != nil { + if err := s.db.Raw("SELECT EXISTS(SELECT 1 FROM "+table+" WHERE token = ?) AS found", nil, tokenstr).Scan(&result).Error; err != nil { if err == gorm.ErrRecordNotFound { return helpers.InputError(e, to.StringPtr("InvalidToken")) } @@ -299,10 +320,11 @@ func New(args *Args) (*Server, error) { IdleTimeout: 5 * time.Minute, } - db, err := gorm.Open(sqlite.Open("cocoon.db"), &gorm.Config{}) + gdb, err := gorm.Open(sqlite.Open("cocoon.db"), &gorm.Config{}) if err != nil { return nil, err } + dbw := db.NewDB(gdb) rkbytes, err := os.ReadFile(args.RotationKeyPath) if err != nil { @@ -341,7 +363,7 @@ func New(args *Args) (*Server, error) { httpd: httpd, echo: e, logger: args.Logger, - db: db, + db: dbw, plcClient: plcClient, privateKey: &pkey, config: &config{ @@ -357,6 +379,9 @@ func New(args *Args) (*Server, error) { }, evtman: events.NewEventManager(events.NewMemPersister()), passport: identity.NewPassport(h, identity.NewMemCache(10_000)), + + dbName: args.DbName, + s3Config: args.S3Config, } s.repoman = NewRepoMan(s) // TODO: this is way too lazy, stop it @@ -461,6 +486,8 @@ func (s *Server) Serve(ctx context.Context) error { } }() + go s.backupRoutine() + for _, relay := range s.config.Relays { cli := xrpc.Client{Host: relay} atproto.SyncRequestCrawl(ctx, &cli, &atproto.SyncRequestCrawl_Input{ @@ -474,3 +501,120 @@ func (s *Server) Serve(ctx context.Context) error { return nil } + +func (s *Server) doBackup() { + start := time.Now() + + s.logger.Info("beginning backup to s3...") + + var buf bytes.Buffer + if err := func() error { + s.logger.Info("reading database bytes...") + s.db.Lock() + defer s.db.Unlock() + + sf, err := os.Open(s.dbName) + if err != nil { + return fmt.Errorf("error opening database for backup: %w", err) + } + defer sf.Close() + + if _, err := io.Copy(&buf, sf); err != nil { + return fmt.Errorf("error reading bytes of backup db: %w", err) + } + + return nil + }(); err != nil { + s.logger.Error("error backing up database", "error", err) + return + } + + if err := func() error { + s.logger.Info("sending to s3...") + + currTime := time.Now().Format("2006-01-02_15-04-05") + key := "cocoon-backup-" + currTime + ".db" + + config := &aws.Config{ + Region: aws.String(s.s3Config.Region), + Credentials: credentials.NewStaticCredentials(s.s3Config.AccessKey, s.s3Config.SecretKey, ""), + } + + if s.s3Config.Endpoint != "" { + config.Endpoint = aws.String(s.s3Config.Endpoint) + config.S3ForcePathStyle = aws.Bool(true) + } + + sess, err := session.NewSession(config) + if err != nil { + return err + } + + svc := s3.New(sess) + + if _, err := svc.PutObject(&s3.PutObjectInput{ + Bucket: aws.String(s.s3Config.Bucket), + Key: aws.String(key), + Body: bytes.NewReader(buf.Bytes()), + }); err != nil { + return fmt.Errorf("error uploading file to s3: %w", err) + } + + s.logger.Info("finished uploading backup to s3", "key", key, "duration", time.Now().Sub(start).Seconds()) + + return nil + }(); err != nil { + s.logger.Error("error uploading database backup", "error", err) + return + } + + os.WriteFile("last-backup.txt", []byte(time.Now().String()), 0644) +} + +func (s *Server) backupRoutine() { + if s.s3Config == nil || !s.s3Config.BackupsEnabled { + return + } + + if s.s3Config.Region == "" { + s.logger.Warn("no s3 region configured but backups are enabled. backups will not run.") + return + } + + if s.s3Config.Bucket == "" { + s.logger.Warn("no s3 bucket configured but backups are enabled. backups will not run.") + return + } + + if s.s3Config.AccessKey == "" { + s.logger.Warn("no s3 access key configured but backups are enabled. backups will not run.") + return + } + + if s.s3Config.SecretKey == "" { + s.logger.Warn("no s3 secret key configured but backups are enabled. backups will not run.") + return + } + + shouldBackupNow := false + lastBackupStr, err := os.ReadFile("last-backup.txt") + if err != nil { + shouldBackupNow = true + } else { + lastBackup, err := time.Parse("2006-01-02 15:04:05.999999999 -0700 MST", string(lastBackupStr)) + if err != nil { + shouldBackupNow = true + } else if time.Now().Sub(lastBackup).Seconds() > 3600 { + shouldBackupNow = true + } + } + + if shouldBackupNow { + go s.doBackup() + } + + ticker := time.NewTicker(time.Hour) + for range ticker.C { + go s.doBackup() + } +} diff --git a/server/session.go b/server/session.go index 4084189..2304cf1 100644 --- a/server/session.go +++ b/server/session.go @@ -55,7 +55,7 @@ func (s *Server) createSession(repo *models.Repo) (*Session, error) { RefreshToken: refreshString, CreatedAt: now, ExpiresAt: accexp, - }).Error; err != nil { + }, nil).Error; err != nil { return nil, err } @@ -64,7 +64,7 @@ func (s *Server) createSession(repo *models.Repo) (*Session, error) { Did: repo.Did, CreatedAt: now, ExpiresAt: refexp, - }).Error; err != nil { + }, nil).Error; err != nil { return nil, err } -- 2.51.2