From 505f7020ad10837616841e9f9f927c16fc202f65 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=A0irhoe=20Biazhkovi=C4=8D?= Date: Mon, 17 Oct 2022 14:08:17 +0300 Subject: [PATCH] (not tested) add named queries support in PG storage * update dependencies --- go.mod | 16 ++--- go.sum | 33 ++++----- storage/pg/storage.go | 133 +++++++++++++++++++++++++------------ storage/pg/storage_test.go | 16 ++--- 4 files changed, 125 insertions(+), 73 deletions(-) diff --git a/go.mod b/go.mod index b9260c4..437ea2a 100644 --- a/go.mod +++ b/go.mod @@ -4,11 +4,11 @@ go 1.18 require ( code.cloudfoundry.org/go-diodes v0.0.0-20220927211948-2985a72b297c - github.com/MicahParks/keyfunc v1.4.0 + github.com/MicahParks/keyfunc v1.5.1 github.com/anacrolix/torrent v1.47.0 github.com/go-redis/redis/v8 v8.11.5 github.com/golang-jwt/jwt/v4 v4.4.2 - github.com/jackc/pgx/v5 v5.0.1 + github.com/jackc/pgx/v5 v5.0.3 github.com/julienschmidt/httprouter v1.3.0 github.com/libp2p/go-reuseport v0.2.0 github.com/minio/sha256-simd v1.0.0 @@ -23,19 +23,19 @@ require ( github.com/anacrolix/dht/v2 v2.19.0 // indirect github.com/anacrolix/log v0.13.2-0.20220711050817-613cb738ef30 // indirect github.com/anacrolix/missinggo v1.3.0 // indirect - github.com/anacrolix/missinggo/v2 v2.7.0 // indirect + github.com/anacrolix/missinggo/v2 v2.7.1 // indirect github.com/beorn7/perks v1.0.1 // indirect github.com/bradfitz/iter v0.0.0-20191230175014-e8f45d346db8 // indirect github.com/cespare/xxhash/v2 v2.1.2 // indirect github.com/davecgh/go-spew v1.1.1 // indirect github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect - github.com/fsnotify/fsnotify v1.5.4 // indirect + github.com/fsnotify/fsnotify v1.6.0 // indirect github.com/golang/protobuf v1.5.2 // indirect github.com/huandu/xstrings v1.3.2 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20200714003250-2b9c44734f2b // indirect github.com/jackc/puddle/v2 v2.0.0 // indirect - github.com/klauspost/cpuid/v2 v2.1.1 // indirect + github.com/klauspost/cpuid/v2 v2.1.2 // indirect github.com/mattn/go-colorable v0.1.13 // indirect github.com/mattn/go-isatty v0.0.16 // indirect github.com/matttproud/golang_protobuf_extensions v1.0.2 // indirect @@ -43,8 +43,8 @@ require ( github.com/prometheus/client_model v0.2.0 // indirect github.com/prometheus/common v0.37.0 // indirect github.com/prometheus/procfs v0.8.0 // indirect - golang.org/x/crypto v0.0.0-20220926161630-eccd6366d1be // indirect - golang.org/x/sys v0.0.0-20220928140112-f11e5e49a4ec // indirect - golang.org/x/text v0.3.7 // indirect + golang.org/x/crypto v0.0.0-20221012134737-56aed061732a // indirect + golang.org/x/sys v0.0.0-20221013171732-95e765b1cc43 // indirect + golang.org/x/text v0.3.8 // indirect google.golang.org/protobuf v1.28.1 // indirect ) diff --git a/go.sum b/go.sum index ad69847..089ed53 100644 --- a/go.sum +++ b/go.sum @@ -37,8 +37,8 @@ crawshaw.io/sqlite v0.3.2/go.mod h1:igAO5JulrQ1DbdZdtVq48mnZUBAPOeFzer7VhDWNtW4= dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo= -github.com/MicahParks/keyfunc v1.4.0 h1:/qTnDiQJSM0pkUvKRVb82jvvm0SAOffAqKSIrgUqwdI= -github.com/MicahParks/keyfunc v1.4.0/go.mod h1:IdnCilugA0O/99dW+/MkvlyrsX8+L8+x95xuVNtM5jw= +github.com/MicahParks/keyfunc v1.5.1 h1:RlyyYgKQI/adkIw1yXYtPvTAOb7hBhSX42aH23d8N0Q= +github.com/MicahParks/keyfunc v1.5.1/go.mod h1:IdnCilugA0O/99dW+/MkvlyrsX8+L8+x95xuVNtM5jw= github.com/RoaringBitmap/roaring v0.4.7/go.mod h1:8khRDP4HmeXns4xIj9oGrKSz7XTQiJx2zgh7AcNke4w= github.com/RoaringBitmap/roaring v0.4.17/go.mod h1:D3qVegWTmfCaX4Bl5CrBE9hfrSrrXIr8KVNvRsDi1NI= github.com/RoaringBitmap/roaring v0.4.23/go.mod h1:D0gp8kJQgE1A4LQ5wFLggQEyvDi06Mq5mKs52e1TwOo= @@ -67,8 +67,8 @@ github.com/anacrolix/missinggo v1.3.0/go.mod h1:bqHm8cE8xr+15uVfMG3BFui/TxyB6//H github.com/anacrolix/missinggo/perf v1.0.0/go.mod h1:ljAFWkBuzkO12MQclXzZrosP5urunoLS0Cbvb4V0uMQ= github.com/anacrolix/missinggo/v2 v2.2.0/go.mod h1:o0jgJoYOyaoYQ4E2ZMISVa9c88BbUBVQQW4QeRkNCGY= github.com/anacrolix/missinggo/v2 v2.5.1/go.mod h1:WEjqh2rmKECd0t1VhQkLGTdIWXO6f6NLjp5GlMZ+6FA= -github.com/anacrolix/missinggo/v2 v2.7.0 h1:4fzOAAn/VCvfWGviLmh64MPMttrlYew81JdPO7nSHvI= -github.com/anacrolix/missinggo/v2 v2.7.0/go.mod h1:2IZIvmRTizALNYFYXsPR7ofXPzJgyBpKZ4kMqMEICkI= +github.com/anacrolix/missinggo/v2 v2.7.1 h1:Y+wL0JC6D2icpwhDpcrRM4THQB/uFcPNYUtZMbYvQgI= +github.com/anacrolix/missinggo/v2 v2.7.1/go.mod h1:2IZIvmRTizALNYFYXsPR7ofXPzJgyBpKZ4kMqMEICkI= github.com/anacrolix/stm v0.2.0/go.mod h1:zoVQRvSiGjGoTmbM0vSLIiaKjWtNPeTvXUSdJQA4hsg= github.com/anacrolix/tagflag v0.0.0-20180109131632-2146c8d41bf0/go.mod h1:1m2U/K6ZT+JZG0+bdMK6qauP49QT4wE5pmhJXOKKCHw= github.com/anacrolix/tagflag v1.0.0/go.mod h1:1m2U/K6ZT+JZG0+bdMK6qauP49QT4wE5pmhJXOKKCHw= @@ -113,8 +113,8 @@ github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1m github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c= github.com/frankban/quicktest v1.14.3 h1:FJKSZTDHjyhriyC81FLQ0LY93eSai0ZyR/ZIkd3ZUKE= github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo= -github.com/fsnotify/fsnotify v1.5.4 h1:jRbGcIw6P2Meqdwuo0H1p6JVLbL5DHKAKlYndzMwVZI= -github.com/fsnotify/fsnotify v1.5.4/go.mod h1:OVB6XrOHzAwXMpEM7uPOzcehqUV2UqJxmVXmkdnm1bU= +github.com/fsnotify/fsnotify v1.6.0 h1:n+5WquG0fcWoWp6xPWfHdbskMCQaFnG6PfBrh1Ky4HY= +github.com/fsnotify/fsnotify v1.6.0/go.mod h1:sl3t1tCWJFWoRz9R8WJCbQihKKwmorjAbSClcnxKAGw= github.com/glycerine/go-unsnap-stream v0.0.0-20180323001048-9f0cb55181dd/go.mod h1:/20jfyN9Y5QPEAprSgKAUr+glWDY39ZiUEAYOEv5dsE= github.com/glycerine/go-unsnap-stream v0.0.0-20181221182339-f9677308dec2/go.mod h1:/20jfyN9Y5QPEAprSgKAUr+glWDY39ZiUEAYOEv5dsE= github.com/glycerine/go-unsnap-stream v0.0.0-20190901134440-81cf024a9e0a/go.mod h1:/20jfyN9Y5QPEAprSgKAUr+glWDY39ZiUEAYOEv5dsE= @@ -215,8 +215,8 @@ github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsI github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= github.com/jackc/pgservicefile v0.0.0-20200714003250-2b9c44734f2b h1:C8S2+VttkHFdOOCXJe+YGfa4vHYwlt4Zx+IVXQ97jYg= github.com/jackc/pgservicefile v0.0.0-20200714003250-2b9c44734f2b/go.mod h1:vsD4gTJCa9TptPL8sPkXrLZ+hDuNrZCnj29CQpr4X1E= -github.com/jackc/pgx/v5 v5.0.1 h1:JZu9othr7l8so2JMDAGeDUMXqERAuZpovyfl4H50tdg= -github.com/jackc/pgx/v5 v5.0.1/go.mod h1:JBbvW3Hdw77jKl9uJrEDATUZIFM2VFPzRq4RWIhkF4o= +github.com/jackc/pgx/v5 v5.0.3 h1:4flM5ecR/555F0EcnjdaZa6MhBU+nr0QbZIo5vaKjuM= +github.com/jackc/pgx/v5 v5.0.3/go.mod h1:JBbvW3Hdw77jKl9uJrEDATUZIFM2VFPzRq4RWIhkF4o= github.com/jackc/puddle/v2 v2.0.0 h1:Kwk/AlLigcnZsDssc3Zun1dk1tAtQNPaBBxBHWn0Mjc= github.com/jackc/puddle/v2 v2.0.0/go.mod h1:itE7ZJY8xnoo0JqJEpSMprN0f+NQkMCuEV/N9j8h0oc= github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4= @@ -234,8 +234,8 @@ github.com/julienschmidt/httprouter v1.3.0 h1:U0609e9tgbseu3rBINet9P48AI/D3oJs4d github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8IZAc4RVcycCCAKdM= github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= github.com/klauspost/cpuid/v2 v2.0.4/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= -github.com/klauspost/cpuid/v2 v2.1.1 h1:t0wUqjowdm8ezddV5k0tLWVklVuvLJpoHeb4WBdydm0= -github.com/klauspost/cpuid/v2 v2.1.1/go.mod h1:RVVoqg1df56z8g3pUjL/3lE5UfnlrJX8tyFgg4nqhuY= +github.com/klauspost/cpuid/v2 v2.1.2 h1:XhdX4fqAJUA0yj+kUwMavO0hHrSPAecYdYf1ZmxHvak= +github.com/klauspost/cpuid/v2 v2.1.2/go.mod h1:RVVoqg1df56z8g3pUjL/3lE5UfnlrJX8tyFgg4nqhuY= github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc= @@ -367,8 +367,8 @@ golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8U golang.org/x/crypto v0.0.0-20190605123033-f99c8df09eb5/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= -golang.org/x/crypto v0.0.0-20220926161630-eccd6366d1be h1:fmw3UbQh+nxngCAHrDCCztao/kbYFnWjoqop8dHx05A= -golang.org/x/crypto v0.0.0-20220926161630-eccd6366d1be/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4= +golang.org/x/crypto v0.0.0-20221012134737-56aed061732a h1:NmSIgad6KjE6VvHciPZuNRTKxGhlPfD6OA87W/PLkqg= +golang.org/x/crypto v0.0.0-20221012134737-56aed061732a/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4= golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8= @@ -492,11 +492,11 @@ golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.0.0-20210927094055-39ccf1dd6fa6/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220114195835-da31bd327af9/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.0.0-20220412211240-33da011f77ad/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220704084225-05e143d24a9e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.0.0-20220928140112-f11e5e49a4ec h1:BkDtF2Ih9xZ7le9ndzTA7KJow28VbQW3odyk/8drmuI= -golang.org/x/sys v0.0.0-20220928140112-f11e5e49a4ec/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20221013171732-95e765b1cc43 h1:OK7RB6t2WQX54srQQYSXMW8dF5C6/8+oA/s5QBmmto4= +golang.org/x/sys v0.0.0-20221013171732-95e765b1cc43/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= @@ -505,8 +505,9 @@ golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= -golang.org/x/text v0.3.7 h1:olpwvP2KacW1ZWvsR7uQhoyTYvKAupfQrRGBFM352Gk= golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= +golang.org/x/text v0.3.8 h1:nAL+RVCQ9uMn3vJZbV+MRnydTJFPf8qqY42YiA6MrqY= +golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ= golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= diff --git a/storage/pg/storage.go b/storage/pg/storage.go index 164bbf8..896222c 100644 --- a/storage/pg/storage.go +++ b/storage/pg/storage.go @@ -35,6 +35,18 @@ const ( errRequiredParameterNotSetMsg = "required parameter not provided: %s" errRequiredColumnsNotFoundMsg = "one or more required columns not found in result set: %v" errRollBackMsg = "error occurred while rolling back failed query: %v, failed query error: %v" + + pCtx = "context" + pKey = "key" + pValue = "value" + pInfoHash = "info_hash" + pPeerID = "peer_id" + pAddress = "address" + pPort = "port" + pV6 = "is_v6" + pSeeder = "is_seeder" + pCreated = "created" + pCount = "count" ) var ( @@ -228,18 +240,13 @@ type store struct { closed chan any } -func (s *store) Put(ctx string, values ...storage.Entry) (err error) { +func (s *store) tx(ctx context.Context, query string, args pgx.NamedArgs) (err error) { var tx pgx.Tx - if tx, err = s.Begin(context.TODO()); err == nil { - for _, v := range values { - if _, err = tx.Exec(context.TODO(), s.Data.AddQuery, ctx, []byte(v.Key), v.Value); err != nil { - break - } - } - if err == nil { - err = tx.Commit(context.TODO()) + if tx, err = s.Begin(ctx); err == nil { + if _, err = tx.Exec(ctx, query, args); err == nil { + err = tx.Commit(ctx) } else { - if txErr := tx.Rollback(context.TODO()); txErr != nil { + if txErr := tx.Rollback(ctx); txErr != nil { err = fmt.Errorf(errRollBackMsg, txErr, err) } } @@ -247,9 +254,39 @@ func (s *store) Put(ctx string, values ...storage.Entry) (err error) { return } +func (s *store) txBatch(ctx context.Context, batch *pgx.Batch) (err error) { + var tx pgx.Tx + if tx, err = s.Begin(ctx); err == nil { + if err = tx.SendBatch(context.TODO(), batch).Close(); err == nil { + err = tx.Commit(ctx) + } else { + if txErr := tx.Rollback(ctx); txErr != nil { + err = fmt.Errorf(errRollBackMsg, txErr, err) + } + } + } + return +} + +func (s *store) Put(ctx string, values ...storage.Entry) (err error) { + switch len(values) { + case 0: + // ignore + case 1: + err = s.tx(context.TODO(), s.Data.AddQuery, pgx.NamedArgs{pCtx: ctx, pKey: []byte(values[0].Key), pValue: values[0].Value}) + default: + var batch pgx.Batch + for _, v := range values { + batch.Queue(s.Data.AddQuery, pgx.NamedArgs{pCtx: ctx, pKey: []byte(v.Key), pValue: v.Value}) + } + err = s.txBatch(context.TODO(), &batch) + } + return +} + func (s *store) Contains(ctx string, key string) (contains bool, err error) { var rows pgx.Rows - if rows, err = s.Query(context.TODO(), s.Data.GetQuery, ctx, []byte(key)); err == nil { + if rows, err = s.Query(context.TODO(), s.Data.GetQuery, pgx.NamedArgs{pCtx: ctx, pKey: []byte(key)}); err == nil { defer rows.Close() contains = rows.Next() err = rows.Err() @@ -258,7 +295,7 @@ func (s *store) Contains(ctx string, key string) (contains bool, err error) { } func (s *store) Load(ctx string, key string) (out []byte, err error) { - row := s.QueryRow(context.TODO(), s.Data.GetQuery, ctx, []byte(key)) + row := s.QueryRow(context.TODO(), s.Data.GetQuery, pgx.NamedArgs{pCtx: ctx, pKey: []byte(key)}) if err = row.Scan(&out); errors.Is(err, pgx.ErrNoRows) { err = nil } @@ -266,20 +303,12 @@ func (s *store) Load(ctx string, key string) (out []byte, err error) { } func (s *store) Delete(ctx string, keys ...string) (err error) { - var tx pgx.Tx - if tx, err = s.Begin(context.TODO()); err == nil { - for _, k := range keys { - if _, err = tx.Exec(context.TODO(), s.Data.DelQuery, ctx, []byte(k)); err != nil { - break - } - } - if err == nil { - err = tx.Commit(context.TODO()) - } else { - if txErr := tx.Rollback(context.TODO()); txErr != nil { - err = fmt.Errorf(errRollBackMsg, txErr, err) - } + if len(keys) > 0 { + baKeys := make([][]byte, len(keys)) + for i, k := range keys { + baKeys[i] = []byte(k) } + _, err = s.Exec(context.TODO(), s.Data.DelQuery, pgx.NamedArgs{pCtx: ctx, pKey: baKeys}) } return } @@ -304,7 +333,7 @@ func (s *store) ScheduleGC(gcInterval, peerLifeTime time.Duration) { return case <-t.C: start := time.Now() - _, err := s.Exec(context.Background(), s.GCQuery, time.Now().Add(-peerLifeTime)) + _, err := s.Exec(context.Background(), s.GCQuery, pgx.NamedArgs{pCreated: time.Now().Add(-peerLifeTime)}) duration := time.Since(start) if err != nil { logger.Error().Err(err).Msg("error occurred while GC") @@ -352,27 +381,40 @@ func (s *store) ScheduleStatisticsCollection(reportInterval time.Duration) { }() } -func (s *store) putPeer(ih bittorrent.InfoHash, peer bittorrent.Peer, seeder bool) error { +func (s *store) putPeer(ih bittorrent.InfoHash, peer bittorrent.Peer, seeder bool) (err error) { logger.Trace(). Stringer("infoHash", ih). Object("peer", peer). Bool("seeder", seeder). Msg("put peer") - args := []any{[]byte(ih), peer.ID[:], net.IP(peer.Addr().AsSlice()), peer.Port(), seeder, peer.Addr().Is6()} - if s.GCAware() { - args = append(args, timecache.Now()) + args := pgx.NamedArgs{ + pInfoHash: []byte(ih), + pPeerID: peer.ID[:], + pAddress: net.IP(peer.Addr().AsSlice()), + pPort: peer.Port(), + pSeeder: seeder, + pV6: peer.Addr().Is6(), } - _, err := s.Exec(context.TODO(), s.Peer.AddQuery, args...) - return err + if s.GCAware() { + args[pCreated] = timecache.Now() + } + _, err = s.Exec(context.TODO(), s.Peer.AddQuery, args) + return } -func (s *store) delPeer(ih bittorrent.InfoHash, peer bittorrent.Peer, seeder bool) error { +func (s *store) delPeer(ih bittorrent.InfoHash, peer bittorrent.Peer, seeder bool) (err error) { logger.Trace(). Stringer("infoHash", ih). Object("peer", peer). Msg("del peer") - _, err := s.Exec(context.TODO(), s.Peer.DelQuery, []byte(ih), peer.ID[:], net.IP(peer.Addr().AsSlice()), peer.Port(), seeder) - return err + _, err = s.Exec(context.TODO(), s.Peer.DelQuery, pgx.NamedArgs{ + pInfoHash: []byte(ih), + pPeerID: peer.ID[:], + pAddress: net.IP(peer.Addr().AsSlice()), + pPort: peer.Port(), + pSeeder: seeder, + }) + return } func (s *store) PutSeeder(ih bittorrent.InfoHash, peer bittorrent.Peer) error { @@ -391,19 +433,28 @@ func (s *store) DeleteLeecher(ih bittorrent.InfoHash, peer bittorrent.Peer) erro return s.delPeer(ih, peer, false) } -func (s *store) GraduateLeecher(ih bittorrent.InfoHash, peer bittorrent.Peer) error { +func (s *store) GraduateLeecher(ih bittorrent.InfoHash, peer bittorrent.Peer) (err error) { logger.Trace(). Stringer("infoHash", ih). Object("peer", peer). Msg("graduate leecher") - _, err := s.Exec(context.TODO(), s.Peer.GraduateQuery, []byte(ih), peer.ID[:], net.IP(peer.Addr().AsSlice()), peer.Port()) - return err + _, err = s.Exec(context.TODO(), s.Peer.GraduateQuery, pgx.NamedArgs{ + pInfoHash: []byte(ih), + pPeerID: peer.ID[:], + pAddress: net.IP(peer.Addr().AsSlice()), + pPort: peer.Port(), + }) + return } func (s *store) getPeers(ih bittorrent.InfoHash, seeders bool, maxCount int, isV6 bool) (peers []bittorrent.Peer, err error) { var rows pgx.Rows - // TODO: see https://github.com/jackc/pgx/issues/387#issuecomment-1107666716 - if rows, err = s.Query(context.TODO(), s.Announce.Query, []byte(ih), seeders, isV6, maxCount); err == nil { + if rows, err = s.Query(context.TODO(), s.Announce.Query, pgx.NamedArgs{ + pInfoHash: []byte(ih), + pSeeder: seeders, + pV6: isV6, + pCount: maxCount, + }); err == nil { defer rows.Close() idIndex, ipIndex, portIndex := -1, -1, -1 for i, field := range rows.FieldDescriptions() { @@ -498,7 +549,7 @@ func (s *store) countPeers(ih bittorrent.InfoHash) (seeders int, leechers int) { if ih == bittorrent.NoneInfoHash { rows, err = s.Query(context.TODO(), s.Peer.CountQuery) } else { - rows, err = s.Query(context.TODO(), s.Peer.CountQuery+" "+s.Peer.ByInfoHashClause, []byte(ih)) + rows, err = s.Query(context.TODO(), s.Peer.CountQuery+" "+s.Peer.ByInfoHashClause, pgx.NamedArgs{pInfoHash: []byte(ih)}) } if err == nil { defer rows.Close() diff --git a/storage/pg/storage_test.go b/storage/pg/storage_test.go index bac4805..e27d084 100644 --- a/storage/pg/storage_test.go +++ b/storage/pg/storage_test.go @@ -40,26 +40,26 @@ var cfg = Config{ ConnectionString: "host=127.0.0.1 database=test user=postgres pool_max_conns=50", PingQuery: "SELECT 1", Peer: peerQueryConf{ - AddQuery: "INSERT INTO mo_peers VALUES($1, $2, $3, $4, $5, $6, $7) ON CONFLICT (info_hash, peer_id, address, port) DO UPDATE SET created = EXCLUDED.created, is_seeder = EXCLUDED.is_seeder", - DelQuery: "DELETE FROM mo_peers WHERE info_hash=$1 AND peer_id=$2 AND address=$3 AND port=$4 AND is_seeder=$5", - GraduateQuery: "UPDATE mo_peers SET is_seeder=TRUE WHERE info_hash=$1 AND peer_id=$2 AND address=$3 AND port=$4 AND NOT is_seeder", + AddQuery: "INSERT INTO mo_peers VALUES(@info_hash, @peer_id, @address, @port, @is_seeder, @is_v6, @created) ON CONFLICT (info_hash, peer_id, address, port) DO UPDATE SET created = EXCLUDED.created, is_seeder = EXCLUDED.is_seeder", + DelQuery: "DELETE FROM mo_peers WHERE info_hash=@info_hash AND peer_id=peer_id AND address=@address AND port=@port AND is_seeder=$5", + GraduateQuery: "UPDATE mo_peers SET is_seeder=TRUE WHERE info_hash=@info_hash AND peer_id=peer_id AND address=@address AND port=@port AND NOT is_seeder", CountQuery: "SELECT COUNT(1) FILTER (WHERE is_seeder) AS seeders, COUNT(1) FILTER (WHERE NOT is_seeder) AS leechers FROM mo_peers", CountSeedersColumn: "seeders", CountLeechersColumn: "leechers", ByInfoHashClause: "WHERE info_hash = $1", }, Announce: announceQueryConf{ - Query: "SELECT peer_id, address, port FROM mo_peers WHERE info_hash=$1 AND is_seeder=$2 AND is_v6=$3 LIMIT $4", + Query: "SELECT peer_id, address, port FROM mo_peers WHERE info_hash=@info_hash AND is_seeder=@is_seeder AND is_v6=@is_v6 LIMIT @count", PeerIDColumn: "peer_id", AddressColumn: "address", PortColumn: "port", }, Data: dataQueryConf{ - AddQuery: "INSERT INTO mo_kv VALUES($1, $2, $3) ON CONFLICT (context, name) DO NOTHING", - GetQuery: "SELECT value FROM mo_kv WHERE context=$1 AND name=$2", - DelQuery: "DELETE FROM mo_kv WHERE context=$1 AND name=$2", + AddQuery: "INSERT INTO mo_kv VALUES(@context, @key, @value) ON CONFLICT (context, name) DO NOTHING", + GetQuery: "SELECT value FROM mo_kv WHERE context=@context AND name=@key", + DelQuery: "DELETE FROM mo_kv WHERE context=@context AND name IN @key", }, - GCQuery: "DELETE FROM mo_peers WHERE created <= $1", + GCQuery: "DELETE FROM mo_peers WHERE created <= @created", InfoHashCountQuery: "SELECT COUNT(DISTINCT info_hash) as info_hashes FROM mo_peers", }