knotmirror: switch to postgres for concurrent writes

j3s.sh / core

forked from tangled.org/core

fork

Configure Feed

Issues Pull Requests Commits Tags

Feed URL

Select the types of activity you want to include in your feed.

Monorepo for Tangled

fork

Configure Feed

Issues Pull Requests Commits Tags

Feed URL

Select the types of activity you want to include in your feed.

knotmirror: switch to postgres for concurrent writes

Signed-off-by: Seongmin Lee <git@boltless.me>

boltless.me 1 month ago 087205bf 07be37da

verified

This commit was signed with the committer's known signature.

boltless.me

SSH Key Fingerprint:

SHA256:2RJEzZAkDasLTx4PI9wgenxutg9oQkR9LGk+9pzW3hM=

+107 -122

11 changed files

expand all collapse all

go.mod

go.sum

knotmirror

adminpage.go

db.go

hosts.go

repos.go

knotmirror.go

knotstream

knotstream.go

resyncer.go

nix

gomod2nix.toml

pkgs

knot-mirror.nix

+3 -3

go.mod

reviewed

··· 33 33 github.com/hiddeco/sshsig v0.2.0 34 34 github.com/hpcloud/tail v1.0.0 35 35 github.com/ipfs/go-cid v0.5.0 36 36 + github.com/jackc/pgx/v5 v5.8.0 36 37 github.com/mattn/go-sqlite3 v1.14.24 37 38 github.com/microcosm-cc/bluemonday v1.0.27 38 39 github.com/openbao/openbao/api/v2 v2.3.0 ··· 160 161 github.com/ipld/go-codec-dagpb v1.7.0 // indirect 161 162 github.com/ipld/go-ipld-prime v0.21.0 // indirect 162 163 github.com/jackc/pgpassfile v1.0.0 // indirect 163 163 - github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect 164 164 - github.com/jackc/pgx/v5 v5.5.0 // indirect 165 165 - github.com/jackc/puddle/v2 v2.2.1 // indirect 164 164 + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect 165 165 + github.com/jackc/puddle/v2 v2.2.2 // indirect 166 166 github.com/jinzhu/inflection v1.0.0 // indirect 167 167 github.com/jinzhu/now v1.1.5 // indirect 168 168 github.com/json-iterator/go v1.1.12 // indirect

+26 -22

go.sum

reviewed

··· 374 374 github.com/ipld/go-ipld-prime v0.21.0/go.mod h1:3RLqy//ERg/y5oShXXdx5YIp50cFGOanyMctpPjsvxQ= 375 375 github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= 376 376 github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= 377 377 - github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk= 378 378 - github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= 379 379 - github.com/jackc/pgx/v5 v5.5.0 h1:NxstgwndsTRy7eq9/kqYc/BZh5w2hHJV86wjvO+1xPw= 380 380 - github.com/jackc/pgx/v5 v5.5.0/go.mod h1:Ig06C2Vu0t5qXC60W8sqIthScaEnFvojjj9dSljmHRA= 381 381 - github.com/jackc/puddle/v2 v2.2.1 h1:RhxXJtFG022u4ibrCSMSiu5aOq1i77R3OHKNJj77OAk= 382 382 - github.com/jackc/puddle/v2 v2.2.1/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= 377 377 + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= 378 378 + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= 379 379 + github.com/jackc/pgx/v5 v5.8.0 h1:TYPDoleBBme0xGSAX3/+NujXXtpZn9HBONkQC7IEZSo= 380 380 + github.com/jackc/pgx/v5 v5.8.0/go.mod h1:QVeDInX2m9VyzvNeiCJVjCkNFqzsNb43204HshNSZKw= 381 381 + github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= 382 382 + github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= 383 383 github.com/jackpal/go-nat-pmp v1.0.2 h1:KzKSgb7qkJvOUTqYl9/Hg/me3pWgBmERKrTGD7BdWus= 384 384 github.com/jackpal/go-nat-pmp v1.0.2/go.mod h1:QPH045xvCAeXUZOxsnwmrtiCoxIr9eob+4orBN1SBKc= 385 385 github.com/jbenet/go-temp-err-catcher v0.1.0 h1:zpb3ZH6wIE8Shj2sKS+khgRvf7T7RABoLk/+KKHggpk= ··· 413 413 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= 414 414 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= 415 415 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= 416 416 + github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= 417 417 + github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= 416 418 github.com/labstack/echo/v4 v4.11.3 h1:Upyu3olaqSHkCjs1EJJwQ3WId8b8b1hxbogyommKktM= 417 419 github.com/labstack/echo/v4 v4.11.3/go.mod h1:UcGuQ8V6ZNRmSweBIJkPvGfwCMIlFmiqrPqiEBfPYws= 418 420 github.com/labstack/gommon v0.4.1 h1:gqEff0p/hTENGMABzezPoPSRtIh1Cvw0ueMOe0/dfOk= ··· 580 582 github.com/polydawn/refmt v0.89.1-0.20221221234430-40501e09de1f/go.mod h1:/zvteZs/GwLtCgZ4BL6CBsk9IKIlexP43ObX9AxTqTw= 581 583 github.com/posthog/posthog-go v1.5.5 h1:2o3j7IrHbTIfxRtj4MPaXKeimuTYg49onNzNBZbwksM= 582 584 github.com/posthog/posthog-go v1.5.5/go.mod h1:3RqUmSnPuwmeVj/GYrS75wNGqcAKdpODiwc83xZWgdE= 583 583 - github.com/prometheus/client_golang v1.22.0 h1:rb93p9lokFEsctTys46VnV1kLCDpVZ0a/Y92Vm0Zc6Q= 584 584 - github.com/prometheus/client_golang v1.22.0/go.mod h1:R7ljNsLXhuQXYZYtw6GAE9AZg8Y7vEW5scdCXrWRXC0= 585 585 + github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o= 586 586 + github.com/prometheus/client_golang v1.23.2/go.mod h1:Tb1a6LWHB3/SPIzCoaDXI4I8UHKeFTEQ1YCr+0Gyqmg= 585 587 github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= 586 588 github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE= 587 587 - github.com/prometheus/common v0.64.0 h1:pdZeA+g617P7oGv1CzdTzyeShxAGrTBsolKNOLQPGO4= 588 588 - github.com/prometheus/common v0.64.0/go.mod h1:0gZns+BLRQ3V6NdaerOhMbwwRbNh9hkGINtQAsP5GS8= 589 589 + github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9ZoGs= 590 590 + github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA= 589 591 github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzMyRg= 590 592 github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is= 591 593 github.com/puzpuzpuz/xsync/v4 v4.2.0 h1:dlxm77dZj2c3rxq0/XNvvUKISAmovoXF4a4qM6Wvkr0= ··· 639 641 github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= 640 642 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= 641 643 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= 642 642 - github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= 643 643 - github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= 644 644 + github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= 645 645 + github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= 644 646 github.com/urfave/cli v1.22.10/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0= 645 647 github.com/urfave/cli/v3 v3.4.1 h1:1M9UOCy5bLmGnuu1yn3t3CB4rG79Rtoxuv1sPhnm6qM= 646 648 github.com/urfave/cli/v3 v3.4.1/go.mod h1:FJSKtM/9AiiTOJL4fJ6TbMUkxBXn7GO9guZqoZtpYpo= ··· 723 725 go.uber.org/zap v1.16.0/go.mod h1:MA8QOfq0BHJwdXa996Y4dYkAqRKB8/1K1QMMZVaNZjQ= 724 726 go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8= 725 727 go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E= 728 728 + go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI= 729 729 + go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU= 726 730 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= 727 731 golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= 728 732 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= ··· 730 734 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= 731 735 golang.org/x/crypto v0.1.0/go.mod h1:RecgLatLF4+eUMCP1PoPZQb+cVrJcOPbHkTkbkB9sbw= 732 736 golang.org/x/crypto v0.19.0/go.mod h1:Iy9bg/ha4yyC70EfRS8jz+B6ybOBKMaSxLj6P6oBDfU= 733 733 - golang.org/x/crypto v0.40.0 h1:r4x+VvoG5Fm+eJcxMaY8CQM7Lb0l1lsmjGBQ6s8BfKM= 734 734 - golang.org/x/crypto v0.40.0/go.mod h1:Qr1vMER5WyS2dfPHAlsOj01wgLbsyWtFn/aY+5+ZdxY= 737 737 + golang.org/x/crypto v0.41.0 h1:WKYxWedPGCTVVl5+WHSSrOBT0O8lx32+zxmHxijgXp4= 738 738 + golang.org/x/crypto v0.41.0/go.mod h1:pO5AFd7FA68rFak7rOAGVuygIISepHftHnr8dr6+sUc= 735 739 golang.org/x/exp v0.0.0-20250620022241-b7579e27df2b h1:M2rDM6z3Fhozi9O7NWsxAkg/yqS/lQJ6PmkyIV3YP+o= 736 740 golang.org/x/exp v0.0.0-20250620022241-b7579e27df2b/go.mod h1:3//PLf8L/X+8b4vuAfHzxeRUl04Adcb341+IGKfnqS8= 737 741 golang.org/x/image v0.31.0 h1:mLChjE2MV6g1S7oqbXC0/UcKijjm5fnJLUYKIYrLESA= ··· 768 772 golang.org/x/net v0.5.0/go.mod h1:DivGGAXEgPSlEBzxGzZI+ZLohi+xUj054jfeKui00ws= 769 773 golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= 770 774 golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg= 771 771 - golang.org/x/net v0.42.0 h1:jzkYrhi3YQWD6MLBJcsklgQsoAcw89EcZbJw8Z614hs= 772 772 - golang.org/x/net v0.42.0/go.mod h1:FF1RA5d3u7nAYA4z2TkclSCKh68eSXtiFwcWQpPXdt8= 775 775 + golang.org/x/net v0.43.0 h1:lat02VYK2j4aLzMzecihNvTlJNQUq316m2Mr9rnM6YE= 776 776 + golang.org/x/net v0.43.0/go.mod h1:vhO1fvI4dGsIjh73sWfUVjj3N7CA9WkKJNQm2svM6Jg= 773 777 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= 774 778 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= 775 779 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= ··· 809 813 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= 810 814 golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= 811 815 golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= 812 812 - golang.org/x/sys v0.34.0 h1:H5Y5sJ2L2JRdyv7ROF1he/lPdvFsd0mJHFw2ThKHxLA= 813 813 - golang.org/x/sys v0.34.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= 816 816 + golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI= 817 817 + golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= 814 818 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= 815 819 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= 816 820 golang.org/x/term v0.1.0/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= ··· 820 824 golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k= 821 825 golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo= 822 826 golang.org/x/term v0.17.0/go.mod h1:lLRBjIVuehSbZlaOtGMbcMncT+aqLLLmKrsjNrUguwk= 823 823 - golang.org/x/term v0.33.0 h1:NuFncQrRcaRvVmgRkvM3j/F00gWIAlcmlB8ACEKmGIg= 824 824 - golang.org/x/term v0.33.0/go.mod h1:s18+ql9tYWp1IfpV9DmCtQDDSRBUjKaw9M1eAv5UeF0= 827 827 + golang.org/x/term v0.34.0 h1:O/2T7POpk0ZZ7MAzMeWFSg6S5IpWd/RXDlM9hgM3DR4= 828 828 + golang.org/x/term v0.34.0/go.mod h1:5jC53AEywhIVebHgPVeg0mj8OD3VO9OzclacVrqpaAw= 825 829 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= 826 830 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= 827 831 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= ··· 876 880 google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= 877 881 google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= 878 882 google.golang.org/protobuf v1.28.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I= 879 879 - google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY= 880 880 - google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY= 883 883 + google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyMc= 884 884 + google.golang.org/protobuf v1.36.8/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU= 881 885 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= 882 886 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= 883 887 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

+7 -25

knotmirror/adminpage.go

reviewed

··· 13 13 "tangled.org/core/appview/pagination" 14 14 "tangled.org/core/knotmirror/db" 15 15 "tangled.org/core/knotmirror/models" 16 16 - "tangled.org/core/orm" 17 16 ) 18 17 19 18 //go:embed templates/*.html ··· 50 49 } 51 50 52 51 func (s *AdminServer) handleRepos() http.HandlerFunc { 53 53 - // TODO: prepare template 54 52 tpl := template.Must(template.New("").Funcs(funcmap()).ParseFS(templateFS, "templates/base.html", "templates/repos.html")) 55 53 return func(w http.ResponseWriter, r *http.Request) { 56 54 pageNum, _ := strconv.Atoi(r.URL.Query().Get("page")) 57 55 if pageNum < 1 { 58 56 pageNum = 1 59 57 } 60 60 - var ( 61 61 - did = r.URL.Query().Get("did") 62 62 - knot = r.URL.Query().Get("knot") 63 63 - state = r.URL.Query().Get("state") 64 64 - ) 65 65 - 66 58 page := pagination.Page{ 67 59 Offset: (pageNum - 1) * repoPageSize, 68 60 Limit: repoPageSize, 69 61 } 70 70 - var filters []orm.Filter 71 62 72 72 - if did != "" { 73 73 - filters = append(filters, orm.FilterEq("did", did)) 74 74 - } 75 75 - if knot != "" { 76 76 - filters = append(filters, orm.FilterEq("knot_domain", knot)) 77 77 - } 78 78 - if state != "" { 79 79 - filters = append(filters, orm.FilterEq("state", state)) 80 80 - } 63 63 + var ( 64 64 + did = r.URL.Query().Get("did") 65 65 + knot = r.URL.Query().Get("knot") 66 66 + state = r.URL.Query().Get("state") 67 67 + ) 81 68 82 82 - repos, err := db.ListRepos(r.Context(), s.db, page, filters...) 69 69 + repos, err := db.ListRepos(r.Context(), s.db, page, did, knot, state) 83 70 if err != nil { 84 71 http.Error(w, err.Error(), http.StatusInternalServerError) 85 72 } ··· 106 93 return func(w http.ResponseWriter, r *http.Request) { 107 94 var status = r.URL.Query().Get("status") 108 95 109 109 - var filters []orm.Filter 110 110 - if status != "" { 111 111 - filters = append(filters, orm.FilterEq("status", status)) 112 112 - } 113 113 - 114 114 - hosts, err := db.ListHosts(r.Context(), s.db, filters...) 96 96 + hosts, err := db.ListHosts(r.Context(), s.db, models.HostStatus(status)) 115 97 if err != nil { 116 98 http.Error(w, err.Error(), http.StatusInternalServerError) 117 99 }

+22 -15

knotmirror/db/db.go

reviewed

··· 4 4 "context" 5 5 "database/sql" 6 6 "fmt" 7 7 - "strings" 7 7 + "time" 8 8 + 8 9 _ "github.com/mattn/go-sqlite3" 10 10 + _ "github.com/jackc/pgx/v5/stdlib" 9 11 ) 10 12 11 11 - func Make(ctx context.Context, dbPath string) (*sql.DB, error) { 12 12 - // https://github.com/mattn/go-sqlite3#connection-string 13 13 - opts := []string{ 14 14 - "_foreign_keys=1", 15 15 - "_journal_mode=WAL", 16 16 - "_synchronous=NORMAL", 17 17 - "_auto_vacuum=incremental", 13 13 + func Make(ctx context.Context, dbUrl string, maxConns int) (*sql.DB, error) { 14 14 + db, err := sql.Open("pgx", dbUrl) 15 15 + if err != nil { 16 16 + return nil, fmt.Errorf("opening db: %w", err) 18 17 } 19 18 20 20 - db, err := sql.Open("sqlite3", dbPath+"?"+strings.Join(opts, "&")) 21 21 - if err != nil { 22 22 - return nil, err 19 19 + db.SetMaxOpenConns(maxConns) 20 20 + db.SetMaxIdleConns(maxConns) 21 21 + db.SetConnMaxIdleTime(time.Hour) 22 22 + 23 23 + pingCtx, cancel := context.WithTimeout(ctx, 5*time.Second) 24 24 + defer cancel() 25 25 + if err := db.PingContext(pingCtx); err != nil { 26 26 + db.Close() 27 27 + return nil, fmt.Errorf("ping db: %w", err) 23 28 } 24 29 25 30 conn, err := db.Conn(ctx) ··· 47 52 retry_count integer not null default 0, 48 53 retry_after integer not null default 0, 49 54 50 50 - unique(did, rkey) 55 55 + constraint repos_pkey primary key (did, rkey) 51 56 ); 52 57 53 58 -- knot hosts 54 59 create table if not exists hosts ( 55 60 hostname text not null, 56 56 - no_ssl integer not null default 0, 61 61 + no_ssl boolean not null default false, 57 62 status text not null default 'active', 58 58 - last_seq integer not null default -1, 63 63 + last_seq bigint not null default -1, 59 64 60 60 - unique(hostname) 65 65 + constraint hosts_pkey primary key (hostname) 61 66 ); 67 67 + 68 68 + create index if not exists idx_repos_aturi on repos (at_uri); 62 69 `) 63 70 if err != nil { 64 71 return nil, fmt.Errorf("initializing db schema: %w", err)

+8 -22

knotmirror/db/hosts.go

reviewed

··· 6 6 "errors" 7 7 "fmt" 8 8 "log" 9 9 - "strings" 10 9 11 10 "tangled.org/core/knotmirror/models" 12 12 - "tangled.org/core/orm" 13 11 ) 14 12 15 13 func UpsertHost(ctx context.Context, e *sql.DB, host *models.Host) error { 16 14 if _, err := e.ExecContext(ctx, 17 15 `insert into hosts (hostname, no_ssl, status, last_seq) 18 18 - values (?, ?, ?, ?) 16 16 + values ($1, $2, $3, $4) 19 17 on conflict(hostname) do update set 20 18 no_ssl = excluded.no_ssl, 21 19 status = excluded.status, ··· 35 33 var host models.Host 36 34 if err := e.QueryRowContext(ctx, 37 35 `select hostname, no_ssl, status, last_seq 38 38 - from hosts where hostname = ?`, 36 36 + from hosts where hostname = $1`, 39 37 hostname, 40 38 ).Scan( 41 39 &host.Hostname, ··· 62 60 continue 63 61 } 64 62 if _, err := tx.ExecContext(ctx, 65 65 - `update hosts set last_seq = ? where hostname = ?`, 63 63 + `update hosts set last_seq = $1 where hostname = $2`, 66 64 cur.LastSeq, 67 65 cur.Hostname, 68 66 ); err != nil { 69 69 - log.Println("failed to persist host cursor", "host:", cur.Hostname, "lastSeq", cur.LastSeq) 67 67 + log.Println("failed to persist host cursor", "host", cur.Hostname, "lastSeq", cur.LastSeq, "err", err) 70 68 } 71 69 } 72 70 return tx.Commit() 73 71 } 74 72 75 75 - func ListHosts(ctx context.Context, e *sql.DB, filters ...orm.Filter) ([]models.Host, error) { 76 76 - var conditions []string 77 77 - var args []any 78 78 - 79 79 - for _, filter := range filters { 80 80 - conditions = append(conditions, filter.Condition()) 81 81 - args = append(args, filter.Arg()...) 82 82 - } 83 83 - 84 84 - whereClause := "" 85 85 - if len(conditions) > 0 { 86 86 - whereClause = " where " + strings.Join(conditions, " and ") 87 87 - } 88 88 - 73 73 + func ListHosts(ctx context.Context, e *sql.DB, status models.HostStatus) ([]models.Host, error) { 89 74 rows, err := e.QueryContext(ctx, 90 75 `select hostname, no_ssl, status, last_seq 91 91 - from hosts` + whereClause, 92 92 - args..., 76 76 + from hosts 77 77 + where status = $1`, 78 78 + status, 93 79 ) 94 80 if err != nil { 95 81 return nil, fmt.Errorf("querying hosts: %w", err)

+24 -17

knotmirror/db/repos.go

reviewed

··· 9 9 "github.com/bluesky-social/indigo/atproto/syntax" 10 10 "tangled.org/core/appview/pagination" 11 11 "tangled.org/core/knotmirror/models" 12 12 - "tangled.org/core/orm" 13 12 ) 14 13 15 14 func AddRepo(ctx context.Context, e *sql.DB, did syntax.DID, rkey syntax.RecordKey, cid syntax.CID, name, knot string) error { 16 15 if _, err := e.ExecContext(ctx, 17 16 `insert into repos (did, rkey, cid, name, knot_domain) 18 18 - values (?, ?, ?, ?, ?)`, 17 17 + values ($1, $2, $3, $4, $5)`, 19 18 did, rkey, cid, name, knot, 20 19 ); err != nil { 21 20 return fmt.Errorf("inserting repo: %w", err) ··· 26 25 func UpsertRepo(ctx context.Context, e *sql.DB, repo *models.Repo) error { 27 26 if _, err := e.ExecContext(ctx, 28 27 `insert into repos (did, rkey, cid, name, knot_domain, git_rev, repo_sha, state, error_msg, retry_count, retry_after) 29 29 - values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) 28 28 + values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) 30 29 on conflict(did, rkey) do update set 31 30 cid = excluded.cid, 32 31 name = excluded.name, ··· 58 57 func UpdateRepoState(ctx context.Context, e *sql.DB, did syntax.DID, rkey syntax.RecordKey, state models.RepoState) error { 59 58 if _, err := e.ExecContext(ctx, 60 59 `update repos 61 61 - set state = ? 62 62 - where did = ? and rkey = ?`, 60 60 + set state = $1 61 61 + where did = $2 and rkey = $3`, 63 62 state, 64 63 did, rkey, 65 64 ); err != nil { ··· 70 69 71 70 func DeleteRepo(ctx context.Context, e *sql.DB, did syntax.DID, rkey syntax.RecordKey) error { 72 71 if _, err := e.ExecContext(ctx, 73 73 - `delete from repos where did = ? and rkey = ?`, 72 72 + `delete from repos where did = $1 and rkey = $2`, 74 73 did, 75 74 rkey, 76 75 ); err != nil { ··· 95 94 retry_count, 96 95 retry_after 97 96 from repos 98 98 - where did = ? and name = ?`, 97 97 + where did = $1 and name = $2`, 99 98 did, 100 99 name, 101 100 ).Scan( ··· 135 134 retry_count, 136 135 retry_after 137 136 from repos 138 138 - where at_uri = ?`, 137 137 + where at_uri = $1`, 139 138 aturi, 140 139 ).Scan( 141 140 &repo.Did, ··· 158 157 return &repo, nil 159 158 } 160 159 161 161 - func ListRepos(ctx context.Context, e *sql.DB, page pagination.Page, filters ...orm.Filter) ([]models.Repo, error) { 160 160 + func ListRepos(ctx context.Context, e *sql.DB, page pagination.Page, did, knot, state string) ([]models.Repo, error) { 162 161 var conditions []string 163 162 var args []any 164 163 165 165 - for _, filter := range filters { 166 166 - conditions = append(conditions, filter.Condition()) 167 167 - args = append(args, filter.Arg()...) 164 164 + pageClause := "" 165 165 + if page.Limit > 0 { 166 166 + pageClause = " limit $1 offset $2 " 167 167 + args = append(args, page.Limit, page.Offset) 168 168 } 169 169 170 170 whereClause := "" 171 171 + if did != "" { 172 172 + conditions = append(conditions, fmt.Sprintf("did = $%d", len(args)+1)) 173 173 + args = append(args, did) 174 174 + } 175 175 + if knot != "" { 176 176 + conditions = append(conditions, fmt.Sprintf("knot_domain = $%d", len(args)+1)) 177 177 + args = append(args, knot) 178 178 + } 179 179 + if state != "" { 180 180 + conditions = append(conditions, fmt.Sprintf("state = $%d", len(args)+1)) 181 181 + args = append(args, state) 182 182 + } 171 183 if len(conditions) > 0 { 172 184 whereClause = "WHERE " + conditions[0] 173 185 for _, condition := range conditions[1:] { 174 186 whereClause += " AND " + condition 175 187 } 176 176 - } 177 177 - pageClause := "" 178 178 - if page.Limit > 0 { 179 179 - pageClause = " limit ? offset ? " 180 180 - args = append(args, page.Limit, page.Offset) 181 188 } 182 189 183 190 query := `

+2 -2

knotmirror/knotmirror.go

reviewed

··· 28 28 29 29 logger.Debug("config loaded:", "config", cfg) 30 30 31 31 - db, err := db.Make(ctx, cfg.DbPath) 31 31 + db, err := db.Make(ctx, cfg.DbPath, 32) 32 32 if err != nil { 33 33 return fmt.Errorf("initializing db: %w", err) 34 34 } 35 35 36 36 res, err := db.ExecContext(ctx, 37 37 - `update repos set state = ? where state = ?`, 37 37 + `update repos set state = $1 where state = $2`, 38 38 models.RepoStateDesynchronized, 39 39 models.RepoStateResyncing, 40 40 )

+1 -2

knotmirror/knotstream/knotstream.go

reviewed

··· 11 11 "tangled.org/core/knotmirror/db" 12 12 "tangled.org/core/knotmirror/models" 13 13 "tangled.org/core/log" 14 14 - "tangled.org/core/orm" 15 14 ) 16 15 17 16 type KnotStream struct { ··· 71 70 } 72 71 73 72 func (s *KnotStream) ResubscribeAllHosts(ctx context.Context) error { 74 74 - hosts, err := db.ListHosts(ctx, s.db, orm.FilterEq("status", "active")) 73 73 + hosts, err := db.ListHosts(ctx, s.db, models.HostStatusActive) 75 74 if err != nil { 76 75 return fmt.Errorf("listing hosts: %w", err) 77 76 }

+8 -4

knotmirror/resyncer.go

reviewed

··· 10 10 "net/url" 11 11 "os" 12 12 "path" 13 13 + "strings" 13 14 "sync" 14 15 "time" 15 16 ··· 87 88 now := time.Now().Unix() 88 89 if err := r.db.QueryRowContext(ctx, 89 90 `update repos 90 90 - set state = ? 91 91 + set state = $1 91 92 where at_uri = ( 92 93 select at_uri from repos 93 93 - where state in (?, ?, ?) 94 94 - and (retry_after = 0 or retry_after < ?) 94 94 + where state in ($2, $3, $4) 95 95 + and (retry_after = 0 or retry_after < $5) 95 96 limit 1 96 97 ) 97 98 returning at_uri ··· 218 219 retryAfter = time.Now().Add(backoff(retryCount, 60) * 60).Unix() 219 220 } 220 221 222 222 + // remove null bytes 223 223 + errMsg = strings.ReplaceAll(errMsg, "\x00", "") 224 224 + 221 225 repo.State = state 222 226 repo.ErrorMsg = errMsg 223 227 repo.RetryCount = retryCount 224 228 repo.RetryAfter = retryAfter 225 229 if dbErr := db.UpsertRepo(ctx, r.db, repo); dbErr != nil { 226 226 - return dbErr 230 230 + return fmt.Errorf("failed to update repo state: %w", err) 227 231 } 228 232 return err 229 233 }

+6 -6

nix/gomod2nix.toml

reviewed

··· 402 402 version = "v1.0.0" 403 403 hash = "sha256-H0nFbC34/3pZUFnuiQk9W7yvAMh6qJDrqvHp+akBPLM=" 404 404 [mod."github.com/jackc/pgservicefile"] 405 405 - version = "v0.0.0-20221227161230-091c0ba34f0a" 406 406 - hash = "sha256-rBtUw15WPPDp2eulHXH5e2zCIed1OPFYwlCpgDOnGRM=" 405 405 + version = "v0.0.0-20240606120523-5a60cdf6a761" 406 406 + hash = "sha256-ETpGsLAA2wcm5xJBayr/mZrCE1YsWbnkbSSX3ptrFn0=" 407 407 [mod."github.com/jackc/pgx/v5"] 408 408 - version = "v5.5.0" 409 409 - hash = "sha256-ifh4kjdg1YFzFaqMf2QDWjUMJerc8B9HH1/h0n7WsIY=" 408 408 + version = "v5.8.0" 409 409 + hash = "sha256-Mq5/A/Obcceu6kKxUv30DPC2ZaVvD8Iq/YtmLm1BVec=" 410 410 [mod."github.com/jackc/puddle/v2"] 411 411 - version = "v2.2.1" 412 412 - hash = "sha256-Edf8SLT/8l+xfHm9IjUGxs1MHtic2VgRyfqb6OzGA9k=" 411 411 + version = "v2.2.2" 412 412 + hash = "sha256-IUxdu4JYfsCh/qlz2SiUWu7EVPHhyooiVA4oaS2Z6yk=" 413 413 [mod."github.com/jinzhu/inflection"] 414 414 version = "v1.0.0" 415 415 hash = "sha256-3h3pHib5MaCXKyKLIMyQnSptDJ16kPjCOQPoEBoQsZg="

-4

nix/pkgs/knot-mirror.nix

reviewed

··· 1 1 { 2 2 buildGoApplication, 3 3 modules, 4 4 - sqlite-lib, 5 4 src, 6 5 }: 7 6 buildGoApplication { ··· 12 11 doCheck = false; 13 12 14 13 subPackages = ["cmd/knotmirror"]; 15 15 - tags = ["libsqlite3"]; 16 14 17 17 - env.CGO_CFLAGS = "-I ${sqlite-lib}/include "; 18 18 - env.CGO_LDFLAGS = "-L ${sqlite-lib}/lib"; 19 15 CGO_ENABLED = 1; 20 16 }