mirror of
https://github.com/sot-tech/mochi.git
synced 2026-07-11 11:08:11 -07:00
(not tested) add named queries support in PG storage
* update dependencies
This commit is contained in:
Širhoe Biazhkovič
@@ -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
|
||||
)
|
||||
|
||||
@@ -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=
|
||||
|
||||
+92
-41
@@ -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()
|
||||
|
||||
@@ -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",
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user