julien.rbrt.fr
forked from
hailey.at/cocoon
Vow, uncensorable PDS written in Go
add metrics, cleanup logger usages (#51)
* cleanup logger usages and add some metrics
* add some basic metrics
hailey.at
+7
cmd/cocoon/main.go
+7
cmd/cocoon/main.go
···
9
9
"os"
10
10
"time"
11
11
12
+
"github.com/bluesky-social/go-util/pkg/telemetry"
12
13
"github.com/bluesky-social/indigo/atproto/atcrypto"
13
14
"github.com/bluesky-social/indigo/atproto/syntax"
14
15
"github.com/haileyok/cocoon/internal/helpers"
···
154
155
Name: "fallback-proxy",
155
156
EnvVars: []string{"COCOON_FALLBACK_PROXY"},
156
157
},
158
+
telemetry.CLIFlagDebug,
159
+
telemetry.CLIFlagMetricsListenAddress,
157
160
},
158
161
Commands: []*cli.Command{
159
162
runServe,
···
177
180
Flags: []cli.Flag{},
178
181
Action: func(cmd *cli.Context) error {
179
182
183
+
logger := telemetry.StartLogger(cmd)
184
+
telemetry.StartMetrics(cmd)
185
+
180
186
s, err := server.New(&server.Args{
187
+
Logger: logger,
181
188
Addr: cmd.String("addr"),
182
189
DbName: cmd.String("db-name"),
183
190
DbType: cmd.String("db-type"),
+30
metrics/metrics.go
+30
metrics/metrics.go
···
1
+
package metrics
2
+
3
+
import (
4
+
"github.com/prometheus/client_golang/prometheus"
5
+
"github.com/prometheus/client_golang/prometheus/promauto"
6
+
)
7
+
8
+
const (
9
+
NAMESPACE = "cocoon"
10
+
)
11
+
12
+
var (
13
+
RelaysConnected = promauto.NewGaugeVec(prometheus.GaugeOpts{
14
+
Namespace: NAMESPACE,
15
+
Name: "relays_connected",
16
+
Help: "number of connected relays, by host",
17
+
}, []string{"host"})
18
+
19
+
RelaySends = promauto.NewCounterVec(prometheus.CounterOpts{
20
+
Namespace: NAMESPACE,
21
+
Name: "relay_sends",
22
+
Help: "number of events sent to a relay, by host",
23
+
}, []string{"host"})
24
+
25
+
RepoOperations = promauto.NewCounterVec(prometheus.CounterOpts{
26
+
Namespace: NAMESPACE,
27
+
Name: "repo_operations",
28
+
Help: "number of operations made against repos",
29
+
}, []string{"kind"})
30
+
)
+1
-1
oauth/dpop/jti_cache.go
+1
-1
oauth/dpop/jti_cache.go
···
14
14
}
15
15
16
16
func newJTICache(size int) *jtiCache {
17
-
cache := cache.NewCache[string, bool]().WithTTL(24 * time.Hour).WithLRU().WithTTL(constants.JTITtl)
17
+
cache := cache.NewCache[string, bool]().WithTTL(24 * time.Hour).WithLRU().WithTTL(constants.JTITtl).WithMaxKeys(size)
18
18
return &jtiCache{
19
19
cache: cache,
20
20
mu: sync.Mutex{},
+2
-1
server/handle_account.go
+2
-1
server/handle_account.go
···
12
12
13
13
func (s *Server) handleAccount(e echo.Context) error {
14
14
ctx := e.Request().Context()
15
+
logger := s.logger.With("name", "handleAuth")
15
16
16
17
repo, sess, err := s.getSessionRepoOrErr(e)
17
18
if err != nil {
···
22
23
23
24
var tokens []provider.OauthToken
24
25
if err := s.db.Raw(ctx, "SELECT * FROM oauth_tokens WHERE sub = ? AND created_at < ? ORDER BY created_at ASC", nil, repo.Repo.Did, oldestPossibleSession).Scan(&tokens).Error; err != nil {
25
-
s.logger.Error("couldnt fetch oauth sessions for account", "did", repo.Repo.Did, "error", err)
26
+
logger.Error("couldnt fetch oauth sessions for account", "did", repo.Repo.Did, "error", err)
26
27
sess.AddFlash("Unable to fetch sessions. See server logs for more details.", "error")
27
28
sess.Save(e.Request(), e.Response())
28
29
return e.Render(200, "account.html", map[string]any{
+3
-2
server/handle_account_revoke.go
+3
-2
server/handle_account_revoke.go
···
11
11
12
12
func (s *Server) handleAccountRevoke(e echo.Context) error {
13
13
ctx := e.Request().Context()
14
+
logger := s.logger.With("name", "handleAcocuntRevoke")
14
15
15
16
var req AccountRevokeInput
16
17
if err := e.Bind(&req); err != nil {
17
-
s.logger.Error("could not bind account revoke request", "error", err)
18
+
logger.Error("could not bind account revoke request", "error", err)
18
19
return helpers.ServerError(e, nil)
19
20
}
20
21
···
24
25
}
25
26
26
27
if err := s.db.Exec(ctx, "DELETE FROM oauth_tokens WHERE sub = ? AND token = ?", nil, repo.Repo.Did, req.Token).Error; err != nil {
27
-
s.logger.Error("couldnt delete oauth session for account", "did", repo.Repo.Did, "token", req.Token, "error", err)
28
+
logger.Error("couldnt delete oauth session for account", "did", repo.Repo.Did, "token", req.Token, "error", err)
28
29
sess.AddFlash("Unable to revoke session. See server logs for more details.", "error")
29
30
sess.Save(e.Request(), e.Response())
30
31
return e.Redirect(303, "/account")
+2
-1
server/handle_account_signin.go
+2
-1
server/handle_account_signin.go
···
63
63
64
64
func (s *Server) handleAccountSigninPost(e echo.Context) error {
65
65
ctx := e.Request().Context()
66
+
logger := s.logger.With("name", "handleAccountSigninPost")
66
67
67
68
var req OauthSigninInput
68
69
if err := e.Bind(&req); err != nil {
69
-
s.logger.Error("error binding sign in req", "error", err)
70
+
logger.Error("error binding sign in req", "error", err)
70
71
return helpers.ServerError(e, nil)
71
72
}
72
73
+4
-2
server/handle_identity_get_recommended_did_credentials.go
+4
-2
server/handle_identity_get_recommended_did_credentials.go
···
8
8
)
9
9
10
10
func (s *Server) handleGetRecommendedDidCredentials(e echo.Context) error {
11
+
logger := s.logger.With("name", "handleIdentityGetRecommendedDidCredentials")
12
+
11
13
repo := e.Get("repo").(*models.RepoActor)
12
14
k, err := atcrypto.ParsePrivateBytesK256(repo.SigningKey)
13
15
if err != nil {
14
-
s.logger.Error("error parsing key", "error", err)
16
+
logger.Error("error parsing key", "error", err)
15
17
return helpers.ServerError(e, nil)
16
18
}
17
19
creds, err := s.plcClient.CreateDidCredentials(k, "", repo.Actor.Handle)
18
20
if err != nil {
19
-
s.logger.Error("error crating did credentials", "error", err)
21
+
logger.Error("error crating did credentials", "error", err)
20
22
return helpers.ServerError(e, nil)
21
23
}
22
24
+3
-2
server/handle_identity_request_plc_operation.go
+3
-2
server/handle_identity_request_plc_operation.go
···
11
11
12
12
func (s *Server) handleIdentityRequestPlcOperationSignature(e echo.Context) error {
13
13
ctx := e.Request().Context()
14
+
logger := s.logger.With("name", "handleIdentityRequestPlcOperationSignature")
14
15
15
16
urepo := e.Get("repo").(*models.RepoActor)
16
17
···
18
19
eat := time.Now().Add(10 * time.Minute).UTC()
19
20
20
21
if err := s.db.Exec(ctx, "UPDATE repos SET plc_operation_code = ?, plc_operation_code_expires_at = ? WHERE did = ?", nil, code, eat, urepo.Repo.Did).Error; err != nil {
21
-
s.logger.Error("error updating user", "error", err)
22
+
logger.Error("error updating user", "error", err)
22
23
return helpers.ServerError(e, nil)
23
24
}
24
25
25
26
if err := s.sendPlcTokenReset(urepo.Email, urepo.Handle, code); err != nil {
26
-
s.logger.Error("error sending mail", "error", err)
27
+
logger.Error("error sending mail", "error", err)
27
28
return helpers.ServerError(e, nil)
28
29
}
29
30
+7
-5
server/handle_identity_sign_plc_operation.go
+7
-5
server/handle_identity_sign_plc_operation.go
···
27
27
}
28
28
29
29
func (s *Server) handleSignPlcOperation(e echo.Context) error {
30
+
logger := s.logger.With("name", "handleSignPlcOperation")
31
+
30
32
repo := e.Get("repo").(*models.RepoActor)
31
33
32
34
var req ComAtprotoSignPlcOperationRequest
33
35
if err := e.Bind(&req); err != nil {
34
-
s.logger.Error("error binding", "error", err)
36
+
logger.Error("error binding", "error", err)
35
37
return helpers.ServerError(e, nil)
36
38
}
37
39
···
54
56
ctx := context.WithValue(e.Request().Context(), "skip-cache", true)
55
57
log, err := identity.FetchDidAuditLog(ctx, nil, repo.Repo.Did)
56
58
if err != nil {
57
-
s.logger.Error("error fetching doc", "error", err)
59
+
logger.Error("error fetching doc", "error", err)
58
60
return helpers.ServerError(e, nil)
59
61
}
60
62
···
83
85
84
86
k, err := atcrypto.ParsePrivateBytesK256(repo.SigningKey)
85
87
if err != nil {
86
-
s.logger.Error("error parsing signing key", "error", err)
88
+
logger.Error("error parsing signing key", "error", err)
87
89
return helpers.ServerError(e, nil)
88
90
}
89
91
90
92
if err := s.plcClient.SignOp(k, &op); err != nil {
91
-
s.logger.Error("error signing plc operation", "error", err)
93
+
logger.Error("error signing plc operation", "error", err)
92
94
return helpers.ServerError(e, nil)
93
95
}
94
96
95
97
if err := s.db.Exec(ctx, "UPDATE repos SET plc_operation_code = NULL, plc_operation_code_expires_at = NULL WHERE did = ?", nil, repo.Repo.Did).Error; err != nil {
96
-
s.logger.Error("error updating repo", "error", err)
98
+
logger.Error("error updating repo", "error", err)
97
99
return helpers.ServerError(e, nil)
98
100
}
99
101
+6
-4
server/handle_identity_submit_plc_operation.go
+6
-4
server/handle_identity_submit_plc_operation.go
···
21
21
}
22
22
23
23
func (s *Server) handleSubmitPlcOperation(e echo.Context) error {
24
+
logger := s.logger.With("name", "handleIdentitySubmitPlcOperation")
25
+
24
26
repo := e.Get("repo").(*models.RepoActor)
25
27
26
28
var req ComAtprotoSubmitPlcOperationRequest
27
29
if err := e.Bind(&req); err != nil {
28
-
s.logger.Error("error binding", "error", err)
30
+
logger.Error("error binding", "error", err)
29
31
return helpers.ServerError(e, nil)
30
32
}
31
33
···
40
42
41
43
k, err := atcrypto.ParsePrivateBytesK256(repo.SigningKey)
42
44
if err != nil {
43
-
s.logger.Error("error parsing key", "error", err)
45
+
logger.Error("error parsing key", "error", err)
44
46
return helpers.ServerError(e, nil)
45
47
}
46
48
required, err := s.plcClient.CreateDidCredentials(k, "", repo.Actor.Handle)
47
49
if err != nil {
48
-
s.logger.Error("error crating did credentials", "error", err)
50
+
logger.Error("error crating did credentials", "error", err)
49
51
return helpers.ServerError(e, nil)
50
52
}
51
53
···
72
74
}
73
75
74
76
if err := s.passport.BustDoc(context.TODO(), repo.Repo.Did); err != nil {
75
-
s.logger.Warn("error busting did doc", "error", err)
77
+
logger.Warn("error busting did doc", "error", err)
76
78
}
77
79
78
80
s.evtman.AddEvent(context.TODO(), &events.XRPCStreamEvent{
+7
-5
server/handle_identity_update_handle.go
+7
-5
server/handle_identity_update_handle.go
···
22
22
}
23
23
24
24
func (s *Server) handleIdentityUpdateHandle(e echo.Context) error {
25
+
logger := s.logger.With("name", "handleIdentityUpdateHandle")
26
+
25
27
repo := e.Get("repo").(*models.RepoActor)
26
28
27
29
var req ComAtprotoIdentityUpdateHandleRequest
28
30
if err := e.Bind(&req); err != nil {
29
-
s.logger.Error("error binding", "error", err)
31
+
logger.Error("error binding", "error", err)
30
32
return helpers.ServerError(e, nil)
31
33
}
32
34
···
41
43
if strings.HasPrefix(repo.Repo.Did, "did:plc:") {
42
44
log, err := identity.FetchDidAuditLog(ctx, nil, repo.Repo.Did)
43
45
if err != nil {
44
-
s.logger.Error("error fetching doc", "error", err)
46
+
logger.Error("error fetching doc", "error", err)
45
47
return helpers.ServerError(e, nil)
46
48
}
47
49
···
68
70
69
71
k, err := atcrypto.ParsePrivateBytesK256(repo.SigningKey)
70
72
if err != nil {
71
-
s.logger.Error("error parsing signing key", "error", err)
73
+
logger.Error("error parsing signing key", "error", err)
72
74
return helpers.ServerError(e, nil)
73
75
}
74
76
···
82
84
}
83
85
84
86
if err := s.passport.BustDoc(context.TODO(), repo.Repo.Did); err != nil {
85
-
s.logger.Warn("error busting did doc", "error", err)
87
+
logger.Warn("error busting did doc", "error", err)
86
88
}
87
89
88
90
s.evtman.AddEvent(context.TODO(), &events.XRPCStreamEvent{
···
95
97
})
96
98
97
99
if err := s.db.Exec(ctx, "UPDATE actors SET handle = ? WHERE did = ?", nil, req.Handle, repo.Repo.Did).Error; err != nil {
98
-
s.logger.Error("error updating handle in db", "error", err)
100
+
logger.Error("error updating handle in db", "error", err)
99
101
return helpers.ServerError(e, nil)
100
102
}
101
103
+11
-10
server/handle_import_repo.go
+11
-10
server/handle_import_repo.go
···
19
19
20
20
func (s *Server) handleRepoImportRepo(e echo.Context) error {
21
21
ctx := e.Request().Context()
22
+
logger := s.logger.With("name", "handleImportRepo")
22
23
23
24
urepo := e.Get("repo").(*models.RepoActor)
24
25
25
26
b, err := io.ReadAll(e.Request().Body)
26
27
if err != nil {
27
-
s.logger.Error("could not read bytes in import request", "error", err)
28
+
logger.Error("could not read bytes in import request", "error", err)
28
29
return helpers.ServerError(e, nil)
29
30
}
30
31
···
32
33
33
34
cs, err := car.NewCarReader(bytes.NewReader(b))
34
35
if err != nil {
35
-
s.logger.Error("could not read car in import request", "error", err)
36
+
logger.Error("could not read car in import request", "error", err)
36
37
return helpers.ServerError(e, nil)
37
38
}
38
39
39
40
orderedBlocks := []blocks.Block{}
40
41
currBlock, err := cs.Next()
41
42
if err != nil {
42
-
s.logger.Error("could not get first block from car", "error", err)
43
+
logger.Error("could not get first block from car", "error", err)
43
44
return helpers.ServerError(e, nil)
44
45
}
45
46
currBlockCt := 1
46
47
47
48
for currBlock != nil {
48
-
s.logger.Info("someone is importing their repo", "block", currBlockCt)
49
+
logger.Info("someone is importing their repo", "block", currBlockCt)
49
50
orderedBlocks = append(orderedBlocks, currBlock)
50
51
next, _ := cs.Next()
51
52
currBlock = next
···
55
56
slices.Reverse(orderedBlocks)
56
57
57
58
if err := bs.PutMany(context.TODO(), orderedBlocks); err != nil {
58
-
s.logger.Error("could not insert blocks", "error", err)
59
+
logger.Error("could not insert blocks", "error", err)
59
60
return helpers.ServerError(e, nil)
60
61
}
61
62
62
63
r, err := repo.OpenRepo(context.TODO(), bs, cs.Header.Roots[0])
63
64
if err != nil {
64
-
s.logger.Error("could not open repo", "error", err)
65
+
logger.Error("could not open repo", "error", err)
65
66
return helpers.ServerError(e, nil)
66
67
}
67
68
···
76
77
cidStr := cid.String()
77
78
b, err := bs.Get(context.TODO(), cid)
78
79
if err != nil {
79
-
s.logger.Error("record bytes don't exist in blockstore", "error", err)
80
+
logger.Error("record bytes don't exist in blockstore", "error", err)
80
81
return helpers.ServerError(e, nil)
81
82
}
82
83
···
96
97
return nil
97
98
}); err != nil {
98
99
tx.Rollback()
99
-
s.logger.Error("record bytes don't exist in blockstore", "error", err)
100
+
logger.Error("record bytes don't exist in blockstore", "error", err)
100
101
return helpers.ServerError(e, nil)
101
102
}
102
103
···
104
105
105
106
root, rev, err := r.Commit(context.TODO(), urepo.SignFor)
106
107
if err != nil {
107
-
s.logger.Error("error committing", "error", err)
108
+
logger.Error("error committing", "error", err)
108
109
return helpers.ServerError(e, nil)
109
110
}
110
111
111
112
if err := s.UpdateRepo(context.TODO(), urepo.Repo.Did, root, rev); err != nil {
112
-
s.logger.Error("error updating repo after commit", "error", err)
113
+
logger.Error("error updating repo after commit", "error", err)
113
114
return helpers.ServerError(e, nil)
114
115
}
115
116
+8
-7
server/handle_oauth_par.go
+8
-7
server/handle_oauth_par.go
···
20
20
21
21
func (s *Server) handleOauthPar(e echo.Context) error {
22
22
ctx := e.Request().Context()
23
+
logger := s.logger.With("name", "handleOauthPar")
23
24
24
25
var parRequest provider.ParRequest
25
26
if err := e.Bind(&parRequest); err != nil {
26
-
s.logger.Error("error binding for par request", "error", err)
27
+
logger.Error("error binding for par request", "error", err)
27
28
return helpers.ServerError(e, nil)
28
29
}
29
30
30
31
if err := e.Validate(parRequest); err != nil {
31
-
s.logger.Error("missing parameters for par request", "error", err)
32
+
logger.Error("missing parameters for par request", "error", err)
32
33
return helpers.InputError(e, nil)
33
34
}
34
35
···
45
46
"error": "use_dpop_nonce",
46
47
})
47
48
}
48
-
s.logger.Error("error getting dpop proof", "error", err)
49
+
logger.Error("error getting dpop proof", "error", err)
49
50
return helpers.InputError(e, nil)
50
51
}
51
52
···
55
56
AllowMissingDpopProof: true,
56
57
})
57
58
if err != nil {
58
-
s.logger.Error("error authenticating client", "client_id", parRequest.ClientID, "error", err)
59
+
logger.Error("error authenticating client", "client_id", parRequest.ClientID, "error", err)
59
60
return helpers.InputError(e, to.StringPtr(err.Error()))
60
61
}
61
62
···
66
67
} else {
67
68
if !client.Metadata.DpopBoundAccessTokens {
68
69
msg := "dpop bound access tokens are not enabled for this client"
69
-
s.logger.Error(msg)
70
+
logger.Error(msg)
70
71
return helpers.InputError(e, &msg)
71
72
}
72
73
73
74
if dpopProof.JKT != *parRequest.DpopJkt {
74
75
msg := "supplied dpop jkt does not match header dpop jkt"
75
-
s.logger.Error(msg)
76
+
logger.Error(msg)
76
77
return helpers.InputError(e, &msg)
77
78
}
78
79
}
···
89
90
}
90
91
91
92
if err := s.db.Create(ctx, authRequest, nil).Error; err != nil {
92
-
s.logger.Error("error creating auth request in db", "error", err)
93
+
logger.Error("error creating auth request in db", "error", err)
93
94
return helpers.ServerError(e, nil)
94
95
}
95
96
+9
-8
server/handle_oauth_token.go
+9
-8
server/handle_oauth_token.go
···
39
39
40
40
func (s *Server) handleOauthToken(e echo.Context) error {
41
41
ctx := e.Request().Context()
42
+
logger := s.logger.With("name", "handleOauthToken")
42
43
43
44
var req OauthTokenRequest
44
45
if err := e.Bind(&req); err != nil {
45
-
s.logger.Error("error binding token request", "error", err)
46
+
logger.Error("error binding token request", "error", err)
46
47
return helpers.ServerError(e, nil)
47
48
}
48
49
···
58
59
"error": "use_dpop_nonce",
59
60
})
60
61
}
61
-
s.logger.Error("error getting dpop proof", "error", err)
62
+
logger.Error("error getting dpop proof", "error", err)
62
63
return helpers.InputError(e, nil)
63
64
}
64
65
···
66
67
AllowMissingDpopProof: true,
67
68
})
68
69
if err != nil {
69
-
s.logger.Error("error authenticating client", "client_id", req.ClientID, "error", err)
70
+
logger.Error("error authenticating client", "client_id", req.ClientID, "error", err)
70
71
return helpers.InputError(e, to.StringPtr(err.Error()))
71
72
}
72
73
···
87
88
var authReq provider.OauthAuthorizationRequest
88
89
// get the lil guy and delete him
89
90
if err := s.db.Raw(ctx, "DELETE FROM oauth_authorization_requests WHERE code = ? RETURNING *", nil, *req.Code).Scan(&authReq).Error; err != nil {
90
-
s.logger.Error("error finding authorization request", "error", err)
91
+
logger.Error("error finding authorization request", "error", err)
91
92
return helpers.ServerError(e, nil)
92
93
}
93
94
···
112
113
case "S256":
113
114
inputChal, err := base64.RawURLEncoding.DecodeString(*authReq.Parameters.CodeChallenge)
114
115
if err != nil {
115
-
s.logger.Error("error decoding code challenge", "error", err)
116
+
logger.Error("error decoding code challenge", "error", err)
116
117
return helpers.ServerError(e, nil)
117
118
}
118
119
···
173
174
RefreshToken: refreshToken,
174
175
Ip: authReq.Ip,
175
176
}, nil).Error; err != nil {
176
-
s.logger.Error("error creating token in db", "error", err)
177
+
logger.Error("error creating token in db", "error", err)
177
178
return helpers.ServerError(e, nil)
178
179
}
179
180
···
202
203
203
204
var oauthToken provider.OauthToken
204
205
if err := s.db.Raw(ctx, "SELECT * FROM oauth_tokens WHERE refresh_token = ?", nil, req.RefreshToken).Scan(&oauthToken).Error; err != nil {
205
-
s.logger.Error("error finding oauth token by refresh token", "error", err, "refresh_token", req.RefreshToken)
206
+
logger.Error("error finding oauth token by refresh token", "error", err, "refresh_token", req.RefreshToken)
206
207
return helpers.ServerError(e, nil)
207
208
}
208
209
···
260
261
}
261
262
262
263
if err := s.db.Exec(ctx, "UPDATE oauth_tokens SET token = ?, refresh_token = ?, expires_at = ?, updated_at = ? WHERE refresh_token = ?", nil, accessString, nextRefreshToken, eat, now, *req.RefreshToken).Error; err != nil {
263
-
s.logger.Error("error updating token", "error", err)
264
+
logger.Error("error updating token", "error", err)
264
265
return helpers.ServerError(e, nil)
265
266
}
266
267
+6
-6
server/handle_proxy.go
+6
-6
server/handle_proxy.go
···
47
47
}
48
48
49
49
func (s *Server) handleProxy(e echo.Context) error {
50
-
lgr := s.logger.With("handler", "handleProxy")
50
+
logger := s.logger.With("handler", "handleProxy")
51
51
52
52
repo, isAuthed := e.Get("repo").(*models.RepoActor)
53
53
···
58
58
59
59
endpoint, svcDid, err := s.getAtprotoProxyEndpointFromRequest(e)
60
60
if err != nil {
61
-
lgr.Error("could not get atproto proxy", "error", err)
61
+
logger.Error("could not get atproto proxy", "error", err)
62
62
return helpers.ServerError(e, nil)
63
63
}
64
64
···
90
90
}
91
91
hj, err := json.Marshal(header)
92
92
if err != nil {
93
-
lgr.Error("error marshaling header", "error", err)
93
+
logger.Error("error marshaling header", "error", err)
94
94
return helpers.ServerError(e, nil)
95
95
}
96
96
···
118
118
}
119
119
pj, err := json.Marshal(payload)
120
120
if err != nil {
121
-
lgr.Error("error marashaling payload", "error", err)
121
+
logger.Error("error marashaling payload", "error", err)
122
122
return helpers.ServerError(e, nil)
123
123
}
124
124
···
129
129
130
130
sk, err := secp256k1secec.NewPrivateKey(repo.SigningKey)
131
131
if err != nil {
132
-
lgr.Error("can't load private key", "error", err)
132
+
logger.Error("can't load private key", "error", err)
133
133
return err
134
134
}
135
135
136
136
R, S, _, err := sk.SignRaw(rand.Reader, hash[:])
137
137
if err != nil {
138
-
lgr.Error("error signing", "error", err)
138
+
logger.Error("error signing", "error", err)
139
139
}
140
140
141
141
rBytes := R.Bytes()
+5
-4
server/handle_repo_apply_writes.go
+5
-4
server/handle_repo_apply_writes.go
···
27
27
28
28
func (s *Server) handleApplyWrites(e echo.Context) error {
29
29
ctx := e.Request().Context()
30
+
logger := s.logger.With("name", "handleRepoApplyWrites")
30
31
31
32
var req ComAtprotoRepoApplyWritesInput
32
33
if err := e.Bind(&req); err != nil {
33
-
s.logger.Error("error binding", "error", err)
34
+
logger.Error("error binding", "error", err)
34
35
return helpers.ServerError(e, nil)
35
36
}
36
37
37
38
if err := e.Validate(req); err != nil {
38
-
s.logger.Error("error validating", "error", err)
39
+
logger.Error("error validating", "error", err)
39
40
return helpers.InputError(e, nil)
40
41
}
41
42
42
43
repo := e.Get("repo").(*models.RepoActor)
43
44
44
45
if repo.Repo.Did != req.Repo {
45
-
s.logger.Warn("mismatched repo/auth")
46
+
logger.Warn("mismatched repo/auth")
46
47
return helpers.InputError(e, nil)
47
48
}
48
49
···
58
59
59
60
results, err := s.repoman.applyWrites(ctx, repo.Repo, ops, req.SwapCommit)
60
61
if err != nil {
61
-
s.logger.Error("error applying writes", "error", err)
62
+
logger.Error("error applying writes", "error", err)
62
63
return helpers.ServerError(e, nil)
63
64
}
64
65
+5
-4
server/handle_repo_create_record.go
+5
-4
server/handle_repo_create_record.go
···
18
18
19
19
func (s *Server) handleCreateRecord(e echo.Context) error {
20
20
ctx := e.Request().Context()
21
+
logger := s.logger.With("name", "handleCreateRecord")
21
22
22
23
repo := e.Get("repo").(*models.RepoActor)
23
24
24
25
var req ComAtprotoRepoCreateRecordInput
25
26
if err := e.Bind(&req); err != nil {
26
-
s.logger.Error("error binding", "error", err)
27
+
logger.Error("error binding", "error", err)
27
28
return helpers.ServerError(e, nil)
28
29
}
29
30
30
31
if err := e.Validate(req); err != nil {
31
-
s.logger.Error("error validating", "error", err)
32
+
logger.Error("error validating", "error", err)
32
33
return helpers.InputError(e, nil)
33
34
}
34
35
35
36
if repo.Repo.Did != req.Repo {
36
-
s.logger.Warn("mismatched repo/auth")
37
+
logger.Warn("mismatched repo/auth")
37
38
return helpers.InputError(e, nil)
38
39
}
39
40
···
53
54
},
54
55
}, req.SwapCommit)
55
56
if err != nil {
56
-
s.logger.Error("error applying writes", "error", err)
57
+
logger.Error("error applying writes", "error", err)
57
58
return helpers.ServerError(e, nil)
58
59
}
59
60
+5
-4
server/handle_repo_delete_record.go
+5
-4
server/handle_repo_delete_record.go
···
16
16
17
17
func (s *Server) handleDeleteRecord(e echo.Context) error {
18
18
ctx := e.Request().Context()
19
+
logger := s.logger.With("name", "handleDeleteRecord")
19
20
20
21
repo := e.Get("repo").(*models.RepoActor)
21
22
22
23
var req ComAtprotoRepoDeleteRecordInput
23
24
if err := e.Bind(&req); err != nil {
24
-
s.logger.Error("error binding", "error", err)
25
+
logger.Error("error binding", "error", err)
25
26
return helpers.ServerError(e, nil)
26
27
}
27
28
28
29
if err := e.Validate(req); err != nil {
29
-
s.logger.Error("error validating", "error", err)
30
+
logger.Error("error validating", "error", err)
30
31
return helpers.InputError(e, nil)
31
32
}
32
33
33
34
if repo.Repo.Did != req.Repo {
34
-
s.logger.Warn("mismatched repo/auth")
35
+
logger.Warn("mismatched repo/auth")
35
36
return helpers.InputError(e, nil)
36
37
}
37
38
···
44
45
},
45
46
}, req.SwapCommit)
46
47
if err != nil {
47
-
s.logger.Error("error applying writes", "error", err)
48
+
logger.Error("error applying writes", "error", err)
48
49
return helpers.ServerError(e, nil)
49
50
}
50
51
+4
-3
server/handle_repo_describe_repo.go
+4
-3
server/handle_repo_describe_repo.go
···
21
21
22
22
func (s *Server) handleDescribeRepo(e echo.Context) error {
23
23
ctx := e.Request().Context()
24
+
logger := s.logger.With("name", "handleDescribeRepo")
24
25
25
26
did := e.QueryParam("repo")
26
27
repo, err := s.getRepoActorByDid(ctx, did)
···
29
30
return helpers.InputError(e, to.StringPtr("RepoNotFound"))
30
31
}
31
32
32
-
s.logger.Error("error looking up repo", "error", err)
33
+
logger.Error("error looking up repo", "error", err)
33
34
return helpers.ServerError(e, nil)
34
35
}
35
36
···
37
38
38
39
diddoc, err := s.passport.FetchDoc(e.Request().Context(), repo.Repo.Did)
39
40
if err != nil {
40
-
s.logger.Error("error fetching diddoc", "error", err)
41
+
logger.Error("error fetching diddoc", "error", err)
41
42
return helpers.ServerError(e, nil)
42
43
}
43
44
···
67
68
68
69
var records []models.Record
69
70
if err := s.db.Raw(ctx, "SELECT DISTINCT(nsid) FROM records WHERE did = ?", nil, repo.Repo.Did).Scan(&records).Error; err != nil {
70
-
s.logger.Error("error getting collections", "error", err)
71
+
logger.Error("error getting collections", "error", err)
71
72
return helpers.ServerError(e, nil)
72
73
}
73
74
+2
-1
server/handle_repo_list_missing_blobs.go
+2
-1
server/handle_repo_list_missing_blobs.go
···
23
23
24
24
func (s *Server) handleListMissingBlobs(e echo.Context) error {
25
25
ctx := e.Request().Context()
26
+
logger := s.logger.With("name", "handleListMissingBlos")
26
27
27
28
urepo := e.Get("repo").(*models.RepoActor)
28
29
···
38
39
39
40
var records []models.Record
40
41
if err := s.db.Raw(ctx, "SELECT * FROM records WHERE did = ?", nil, urepo.Repo.Did).Scan(&records).Error; err != nil {
41
-
s.logger.Error("failed to get records for listMissingBlobs", "error", err)
42
+
logger.Error("failed to get records for listMissingBlobs", "error", err)
42
43
return helpers.ServerError(e, nil)
43
44
}
44
45
+3
-2
server/handle_repo_list_records.go
+3
-2
server/handle_repo_list_records.go
···
47
47
48
48
func (s *Server) handleListRecords(e echo.Context) error {
49
49
ctx := e.Request().Context()
50
+
logger := s.logger.With("name", "handleListRecords")
50
51
51
52
var req ComAtprotoRepoListRecordsRequest
52
53
if err := e.Bind(&req); err != nil {
53
-
s.logger.Error("could not bind list records request", "error", err)
54
+
logger.Error("could not bind list records request", "error", err)
54
55
return helpers.ServerError(e, nil)
55
56
}
56
57
···
96
97
97
98
var records []models.Record
98
99
if err := s.db.Raw(ctx, "SELECT * FROM records WHERE did = ? AND nsid = ? "+cursorquery+" ORDER BY created_at "+sort+" limit ?", nil, params...).Scan(&records).Error; err != nil {
99
-
s.logger.Error("error getting records", "error", err)
100
+
logger.Error("error getting records", "error", err)
100
101
return helpers.ServerError(e, nil)
101
102
}
102
103
+5
-4
server/handle_repo_put_record.go
+5
-4
server/handle_repo_put_record.go
···
18
18
19
19
func (s *Server) handlePutRecord(e echo.Context) error {
20
20
ctx := e.Request().Context()
21
+
logger := s.logger.With("name", "handlePutRecord")
21
22
22
23
repo := e.Get("repo").(*models.RepoActor)
23
24
24
25
var req ComAtprotoRepoPutRecordInput
25
26
if err := e.Bind(&req); err != nil {
26
-
s.logger.Error("error binding", "error", err)
27
+
logger.Error("error binding", "error", err)
27
28
return helpers.ServerError(e, nil)
28
29
}
29
30
30
31
if err := e.Validate(req); err != nil {
31
-
s.logger.Error("error validating", "error", err)
32
+
logger.Error("error validating", "error", err)
32
33
return helpers.InputError(e, nil)
33
34
}
34
35
35
36
if repo.Repo.Did != req.Repo {
36
-
s.logger.Warn("mismatched repo/auth")
37
+
logger.Warn("mismatched repo/auth")
37
38
return helpers.InputError(e, nil)
38
39
}
39
40
···
53
54
},
54
55
}, req.SwapCommit)
55
56
if err != nil {
56
-
s.logger.Error("error applying writes", "error", err)
57
+
logger.Error("error applying writes", "error", err)
57
58
return helpers.ServerError(e, nil)
58
59
}
59
60
+8
-7
server/handle_repo_upload_blob.go
+8
-7
server/handle_repo_upload_blob.go
···
33
33
34
34
func (s *Server) handleRepoUploadBlob(e echo.Context) error {
35
35
ctx := e.Request().Context()
36
+
logger := s.logger.With("name", "handleRepoUploadBlob")
36
37
37
38
urepo := e.Get("repo").(*models.RepoActor)
38
39
···
54
55
}
55
56
56
57
if err := s.db.Create(ctx, &blob, nil).Error; err != nil {
57
-
s.logger.Error("error creating new blob in db", "error", err)
58
+
logger.Error("error creating new blob in db", "error", err)
58
59
return helpers.ServerError(e, nil)
59
60
}
60
61
···
71
72
break
72
73
}
73
74
} else if err != nil && err != io.ErrUnexpectedEOF {
74
-
s.logger.Error("error reading blob", "error", err)
75
+
logger.Error("error reading blob", "error", err)
75
76
return helpers.ServerError(e, nil)
76
77
}
77
78
···
87
88
}
88
89
89
90
if err := s.db.Create(ctx, &blobPart, nil).Error; err != nil {
90
-
s.logger.Error("error adding blob part to db", "error", err)
91
+
logger.Error("error adding blob part to db", "error", err)
91
92
return helpers.ServerError(e, nil)
92
93
}
93
94
}
···
100
101
101
102
c, err := cid.NewPrefixV1(cid.Raw, multihash.SHA2_256).Sum(fulldata.Bytes())
102
103
if err != nil {
103
-
s.logger.Error("error creating cid prefix", "error", err)
104
+
logger.Error("error creating cid prefix", "error", err)
104
105
return helpers.ServerError(e, nil)
105
106
}
106
107
···
117
118
118
119
sess, err := session.NewSession(config)
119
120
if err != nil {
120
-
s.logger.Error("error creating aws session", "error", err)
121
+
logger.Error("error creating aws session", "error", err)
121
122
return helpers.ServerError(e, nil)
122
123
}
123
124
···
128
129
Key: aws.String(fmt.Sprintf("blobs/%s/%s", urepo.Repo.Did, c.String())),
129
130
Body: bytes.NewReader(fulldata.Bytes()),
130
131
}); err != nil {
131
-
s.logger.Error("error uploading blob to s3", "error", err)
132
+
logger.Error("error uploading blob to s3", "error", err)
132
133
return helpers.ServerError(e, nil)
133
134
}
134
135
}
135
136
136
137
if err := s.db.Exec(ctx, "UPDATE blobs SET cid = ? WHERE id = ?", nil, c.Bytes(), blob.ID).Error; err != nil {
137
138
// there should probably be somme handling here if this fails...
138
-
s.logger.Error("error updating blob", "error", err)
139
+
logger.Error("error updating blob", "error", err)
139
140
return helpers.ServerError(e, nil)
140
141
}
141
142
+3
-2
server/handle_server_activate_account.go
+3
-2
server/handle_server_activate_account.go
···
19
19
20
20
func (s *Server) handleServerActivateAccount(e echo.Context) error {
21
21
ctx := e.Request().Context()
22
+
logger := s.logger.With("name", "handleServerActivateAccount")
22
23
23
24
var req ComAtprotoServerDeactivateAccountRequest
24
25
if err := e.Bind(&req); err != nil {
25
-
s.logger.Error("error binding", "error", err)
26
+
logger.Error("error binding", "error", err)
26
27
return helpers.ServerError(e, nil)
27
28
}
28
29
29
30
urepo := e.Get("repo").(*models.RepoActor)
30
31
31
32
if err := s.db.Exec(ctx, "UPDATE repos SET deactivated = ? WHERE did = ?", nil, false, urepo.Repo.Did).Error; err != nil {
32
-
s.logger.Error("error updating account status to deactivated", "error", err)
33
+
logger.Error("error updating account status to deactivated", "error", err)
33
34
return helpers.ServerError(e, nil)
34
35
}
35
36
+5
-4
server/handle_server_check_account_status.go
+5
-4
server/handle_server_check_account_status.go
···
21
21
22
22
func (s *Server) handleServerCheckAccountStatus(e echo.Context) error {
23
23
ctx := e.Request().Context()
24
+
logger := s.logger.With("name", "handleServerCheckAccountStatus")
24
25
25
26
urepo := e.Get("repo").(*models.RepoActor)
26
27
···
33
34
34
35
rootcid, err := cid.Cast(urepo.Root)
35
36
if err != nil {
36
-
s.logger.Error("error casting cid", "error", err)
37
+
logger.Error("error casting cid", "error", err)
37
38
return helpers.ServerError(e, nil)
38
39
}
39
40
resp.RepoCommit = rootcid.String()
···
44
45
45
46
var blockCtResp CountResp
46
47
if err := s.db.Raw(ctx, "SELECT COUNT(*) AS ct FROM blocks WHERE did = ?", nil, urepo.Repo.Did).Scan(&blockCtResp).Error; err != nil {
47
-
s.logger.Error("error getting block count", "error", err)
48
+
logger.Error("error getting block count", "error", err)
48
49
return helpers.ServerError(e, nil)
49
50
}
50
51
resp.RepoBlocks = blockCtResp.Ct
51
52
52
53
var recCtResp CountResp
53
54
if err := s.db.Raw(ctx, "SELECT COUNT(*) AS ct FROM records WHERE did = ?", nil, urepo.Repo.Did).Scan(&recCtResp).Error; err != nil {
54
-
s.logger.Error("error getting record count", "error", err)
55
+
logger.Error("error getting record count", "error", err)
55
56
return helpers.ServerError(e, nil)
56
57
}
57
58
resp.IndexedRecords = recCtResp.Ct
58
59
59
60
var blobCtResp CountResp
60
61
if err := s.db.Raw(ctx, "SELECT COUNT(*) AS ct FROM blobs WHERE did = ?", nil, urepo.Repo.Did).Scan(&blobCtResp).Error; err != nil {
61
-
s.logger.Error("error getting record count", "error", err)
62
+
logger.Error("error getting record count", "error", err)
62
63
return helpers.ServerError(e, nil)
63
64
}
64
65
resp.ExpectedBlobs = blobCtResp.Ct
+3
-2
server/handle_server_confirm_email.go
+3
-2
server/handle_server_confirm_email.go
···
16
16
17
17
func (s *Server) handleServerConfirmEmail(e echo.Context) error {
18
18
ctx := e.Request().Context()
19
+
logger := s.logger.With("name", "handleServerConfirmEmail")
19
20
20
21
urepo := e.Get("repo").(*models.RepoActor)
21
22
22
23
var req ComAtprotoServerConfirmEmailRequest
23
24
if err := e.Bind(&req); err != nil {
24
-
s.logger.Error("error binding", "error", err)
25
+
logger.Error("error binding", "error", err)
25
26
return helpers.ServerError(e, nil)
26
27
}
27
28
···
44
45
now := time.Now().UTC()
45
46
46
47
if err := s.db.Exec(ctx, "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 {
47
-
s.logger.Error("error updating user", "error", err)
48
+
logger.Error("error updating user", "error", err)
48
49
return helpers.ServerError(e, nil)
49
50
}
50
51
+23
-22
server/handle_server_create_account.go
+23
-22
server/handle_server_create_account.go
···
37
37
38
38
func (s *Server) handleCreateAccount(e echo.Context) error {
39
39
ctx := e.Request().Context()
40
+
logger := s.logger.With("name", "handleServerCreateAccount")
40
41
41
42
var request ComAtprotoServerCreateAccountRequest
42
43
43
44
if err := e.Bind(&request); err != nil {
44
-
s.logger.Error("error receiving request", "endpoint", "com.atproto.server.createAccount", "error", err)
45
+
logger.Error("error receiving request", "endpoint", "com.atproto.server.createAccount", "error", err)
45
46
return helpers.ServerError(e, nil)
46
47
}
47
48
48
49
request.Handle = strings.ToLower(request.Handle)
49
50
50
51
if err := e.Validate(request); err != nil {
51
-
s.logger.Error("error validating request", "endpoint", "com.atproto.server.createAccount", "error", err)
52
+
logger.Error("error validating request", "endpoint", "com.atproto.server.createAccount", "error", err)
52
53
53
54
var verr ValidationError
54
55
if errors.As(err, &verr) {
···
82
83
authDid, err := s.validateServiceAuth(e.Request().Context(), token, "com.atproto.server.createAccount")
83
84
84
85
if err != nil {
85
-
s.logger.Warn("error validating authorization token", "endpoint", "com.atproto.server.createAccount", "error", err)
86
+
logger.Warn("error validating authorization token", "endpoint", "com.atproto.server.createAccount", "error", err)
86
87
return helpers.UnauthorizedError(e, to.StringPtr("invalid authorization token"))
87
88
}
88
89
···
94
95
// see if the handle is already taken
95
96
actor, err := s.getActorByHandle(ctx, request.Handle)
96
97
if err != nil && err != gorm.ErrRecordNotFound {
97
-
s.logger.Error("error looking up handle in db", "endpoint", "com.atproto.server.createAccount", "error", err)
98
+
logger.Error("error looking up handle in db", "endpoint", "com.atproto.server.createAccount", "error", err)
98
99
return helpers.ServerError(e, nil)
99
100
}
100
101
if err == nil && actor.Did != signupDid {
···
115
116
if err == gorm.ErrRecordNotFound {
116
117
return helpers.InputError(e, to.StringPtr("InvalidInviteCode"))
117
118
}
118
-
s.logger.Error("error getting invite code from db", "error", err)
119
+
logger.Error("error getting invite code from db", "error", err)
119
120
return helpers.ServerError(e, nil)
120
121
}
121
122
···
127
128
// see if the email is already taken
128
129
existingRepo, err := s.getRepoByEmail(ctx, request.Email)
129
130
if err != nil && err != gorm.ErrRecordNotFound {
130
-
s.logger.Error("error looking up email in db", "endpoint", "com.atproto.server.createAccount", "error", err)
131
+
logger.Error("error looking up email in db", "endpoint", "com.atproto.server.createAccount", "error", err)
131
132
return helpers.ServerError(e, nil)
132
133
}
133
134
if err == nil && existingRepo.Did != signupDid {
···
141
142
if signupDid != "" {
142
143
reservedKey, err := s.getReservedKey(ctx, signupDid)
143
144
if err != nil {
144
-
s.logger.Error("error looking up reserved key", "error", err)
145
+
logger.Error("error looking up reserved key", "error", err)
145
146
}
146
147
if reservedKey != nil {
147
148
k, err = atcrypto.ParsePrivateBytesK256(reservedKey.PrivateKey)
148
149
if err != nil {
149
-
s.logger.Error("error parsing reserved key", "error", err)
150
+
logger.Error("error parsing reserved key", "error", err)
150
151
k = nil
151
152
} else {
152
153
defer func() {
153
154
if delErr := s.deleteReservedKey(ctx, reservedKey.KeyDid, reservedKey.Did); delErr != nil {
154
-
s.logger.Error("error deleting reserved key", "error", delErr)
155
+
logger.Error("error deleting reserved key", "error", delErr)
155
156
}
156
157
}()
157
158
}
···
161
162
if k == nil {
162
163
k, err = atcrypto.GeneratePrivateKeyK256()
163
164
if err != nil {
164
-
s.logger.Error("error creating signing key", "endpoint", "com.atproto.server.createAccount", "error", err)
165
+
logger.Error("error creating signing key", "endpoint", "com.atproto.server.createAccount", "error", err)
165
166
return helpers.ServerError(e, nil)
166
167
}
167
168
}
···
169
170
if signupDid == "" {
170
171
did, op, err := s.plcClient.CreateDID(k, "", request.Handle)
171
172
if err != nil {
172
-
s.logger.Error("error creating operation", "endpoint", "com.atproto.server.createAccount", "error", err)
173
+
logger.Error("error creating operation", "endpoint", "com.atproto.server.createAccount", "error", err)
173
174
return helpers.ServerError(e, nil)
174
175
}
175
176
176
177
if err := s.plcClient.SendOperation(e.Request().Context(), did, op); err != nil {
177
-
s.logger.Error("error sending plc op", "endpoint", "com.atproto.server.createAccount", "error", err)
178
+
logger.Error("error sending plc op", "endpoint", "com.atproto.server.createAccount", "error", err)
178
179
return helpers.ServerError(e, nil)
179
180
}
180
181
signupDid = did
···
182
183
183
184
hashed, err := bcrypt.GenerateFromPassword([]byte(request.Password), 10)
184
185
if err != nil {
185
-
s.logger.Error("error hashing password", "error", err)
186
+
logger.Error("error hashing password", "error", err)
186
187
return helpers.ServerError(e, nil)
187
188
}
188
189
···
202
203
}
203
204
204
205
if err := s.db.Create(ctx, &urepo, nil).Error; err != nil {
205
-
s.logger.Error("error inserting new repo", "error", err)
206
+
logger.Error("error inserting new repo", "error", err)
206
207
return helpers.ServerError(e, nil)
207
208
}
208
209
209
210
if err := s.db.Create(ctx, &actor, nil).Error; err != nil {
210
-
s.logger.Error("error inserting new actor", "error", err)
211
+
logger.Error("error inserting new actor", "error", err)
211
212
return helpers.ServerError(e, nil)
212
213
}
213
214
} else {
214
215
if err := s.db.Save(ctx, &actor, nil).Error; err != nil {
215
-
s.logger.Error("error inserting new actor", "error", err)
216
+
logger.Error("error inserting new actor", "error", err)
216
217
return helpers.ServerError(e, nil)
217
218
}
218
219
}
···
223
224
224
225
root, rev, err := r.Commit(context.TODO(), urepo.SignFor)
225
226
if err != nil {
226
-
s.logger.Error("error committing", "error", err)
227
+
logger.Error("error committing", "error", err)
227
228
return helpers.ServerError(e, nil)
228
229
}
229
230
230
231
if err := s.UpdateRepo(context.TODO(), urepo.Did, root, rev); err != nil {
231
-
s.logger.Error("error updating repo after commit", "error", err)
232
+
logger.Error("error updating repo after commit", "error", err)
232
233
return helpers.ServerError(e, nil)
233
234
}
234
235
···
244
245
245
246
if s.config.RequireInvite {
246
247
if err := s.db.Raw(ctx, "UPDATE invite_codes SET remaining_use_count = remaining_use_count - 1 WHERE code = ?", nil, request.InviteCode).Scan(&ic).Error; err != nil {
247
-
s.logger.Error("error decrementing use count", "error", err)
248
+
logger.Error("error decrementing use count", "error", err)
248
249
return helpers.ServerError(e, nil)
249
250
}
250
251
}
251
252
252
253
sess, err := s.createSession(ctx, &urepo)
253
254
if err != nil {
254
-
s.logger.Error("error creating new session", "error", err)
255
+
logger.Error("error creating new session", "error", err)
255
256
return helpers.ServerError(e, nil)
256
257
}
257
258
258
259
go func() {
259
260
if err := s.sendEmailVerification(urepo.Email, actor.Handle, *urepo.EmailVerificationCode); err != nil {
260
-
s.logger.Error("error sending email verification email", "error", err)
261
+
logger.Error("error sending email verification email", "error", err)
261
262
}
262
263
if err := s.sendWelcomeMail(urepo.Email, actor.Handle); err != nil {
263
-
s.logger.Error("error sending welcome email", "error", err)
264
+
logger.Error("error sending welcome email", "error", err)
264
265
}
265
266
}()
266
267
+4
-3
server/handle_server_create_invite_code.go
+4
-3
server/handle_server_create_invite_code.go
···
18
18
19
19
func (s *Server) handleCreateInviteCode(e echo.Context) error {
20
20
ctx := e.Request().Context()
21
+
logger := s.logger.With("name", "handleServerCreateInviteCode")
21
22
22
23
var req ComAtprotoServerCreateInviteCodeRequest
23
24
if err := e.Bind(&req); err != nil {
24
-
s.logger.Error("error binding", "error", err)
25
+
logger.Error("error binding", "error", err)
25
26
return helpers.ServerError(e, nil)
26
27
}
27
28
28
29
if err := e.Validate(req); err != nil {
29
-
s.logger.Error("error validating", "error", err)
30
+
logger.Error("error validating", "error", err)
30
31
return helpers.InputError(e, nil)
31
32
}
32
33
···
44
45
Did: acc,
45
46
RemainingUseCount: req.UseCount,
46
47
}, nil).Error; err != nil {
47
-
s.logger.Error("error creating invite code", "error", err)
48
+
logger.Error("error creating invite code", "error", err)
48
49
return helpers.ServerError(e, nil)
49
50
}
50
51
+4
-3
server/handle_server_create_invite_codes.go
+4
-3
server/handle_server_create_invite_codes.go
···
23
23
24
24
func (s *Server) handleCreateInviteCodes(e echo.Context) error {
25
25
ctx := e.Request().Context()
26
+
logger := s.logger.With("name", "handleServerCreateInviteCodes")
26
27
27
28
var req ComAtprotoServerCreateInviteCodesRequest
28
29
if err := e.Bind(&req); err != nil {
29
-
s.logger.Error("error binding", "error", err)
30
+
logger.Error("error binding", "error", err)
30
31
return helpers.ServerError(e, nil)
31
32
}
32
33
33
34
if err := e.Validate(req); err != nil {
34
-
s.logger.Error("error validating", "error", err)
35
+
logger.Error("error validating", "error", err)
35
36
return helpers.InputError(e, nil)
36
37
}
37
38
···
57
58
Did: did,
58
59
RemainingUseCount: req.UseCount,
59
60
}, nil).Error; err != nil {
60
-
s.logger.Error("error creating invite code", "error", err)
61
+
logger.Error("error creating invite code", "error", err)
61
62
return helpers.ServerError(e, nil)
62
63
}
63
64
}
+5
-4
server/handle_server_create_session.go
+5
-4
server/handle_server_create_session.go
···
33
33
34
34
func (s *Server) handleCreateSession(e echo.Context) error {
35
35
ctx := e.Request().Context()
36
+
logger := s.logger.With("name", "handleServerCreateSession")
36
37
37
38
var req ComAtprotoServerCreateSessionRequest
38
39
if err := e.Bind(&req); err != nil {
39
-
s.logger.Error("error binding request", "endpoint", "com.atproto.server.serverCreateSession", "error", err)
40
+
logger.Error("error binding request", "endpoint", "com.atproto.server.serverCreateSession", "error", err)
40
41
return helpers.ServerError(e, nil)
41
42
}
42
43
···
79
80
return helpers.InputError(e, to.StringPtr("InvalidRequest"))
80
81
}
81
82
82
-
s.logger.Error("erorr looking up repo", "endpoint", "com.atproto.server.createSession", "error", err)
83
+
logger.Error("erorr looking up repo", "endpoint", "com.atproto.server.createSession", "error", err)
83
84
return helpers.ServerError(e, nil)
84
85
}
85
86
86
87
if err := bcrypt.CompareHashAndPassword([]byte(repo.Password), []byte(req.Password)); err != nil {
87
88
if err != bcrypt.ErrMismatchedHashAndPassword {
88
-
s.logger.Error("erorr comparing hash and password", "error", err)
89
+
logger.Error("erorr comparing hash and password", "error", err)
89
90
}
90
91
return helpers.InputError(e, to.StringPtr("InvalidRequest"))
91
92
}
92
93
93
94
sess, err := s.createSession(ctx, &repo.Repo)
94
95
if err != nil {
95
-
s.logger.Error("error creating session", "error", err)
96
+
logger.Error("error creating session", "error", err)
96
97
return helpers.ServerError(e, nil)
97
98
}
98
99
+3
-2
server/handle_server_deactivate_account.go
+3
-2
server/handle_server_deactivate_account.go
···
20
20
21
21
func (s *Server) handleServerDeactivateAccount(e echo.Context) error {
22
22
ctx := e.Request().Context()
23
+
logger := s.logger.With("name", "handleServerDeactivateAccount")
23
24
24
25
var req ComAtprotoServerDeactivateAccountRequest
25
26
if err := e.Bind(&req); err != nil {
26
-
s.logger.Error("error binding", "error", err)
27
+
logger.Error("error binding", "error", err)
27
28
return helpers.ServerError(e, nil)
28
29
}
29
30
30
31
urepo := e.Get("repo").(*models.RepoActor)
31
32
32
33
if err := s.db.Exec(ctx, "UPDATE repos SET deactivated = ? WHERE did = ?", nil, true, urepo.Repo.Did).Error; err != nil {
33
-
s.logger.Error("error updating account status to deactivated", "error", err)
34
+
logger.Error("error updating account status to deactivated", "error", err)
34
35
return helpers.ServerError(e, nil)
35
36
}
36
37
+30
-27
server/handle_server_delete_account.go
+30
-27
server/handle_server_delete_account.go
···
21
21
22
22
func (s *Server) handleServerDeleteAccount(e echo.Context) error {
23
23
ctx := e.Request().Context()
24
+
logger := s.logger.With("name", "handleServerDeleteAccount")
24
25
25
26
var req ComAtprotoServerDeleteAccountRequest
26
27
if err := e.Bind(&req); err != nil {
27
-
s.logger.Error("error binding", "error", err)
28
+
logger.Error("error binding", "error", err)
28
29
return helpers.ServerError(e, nil)
29
30
}
30
31
31
32
if err := e.Validate(&req); err != nil {
32
-
s.logger.Error("error validating", "error", err)
33
+
logger.Error("error validating", "error", err)
33
34
return helpers.ServerError(e, nil)
34
35
}
35
36
36
37
urepo, err := s.getRepoActorByDid(ctx, req.Did)
37
38
if err != nil {
38
-
s.logger.Error("error getting repo", "error", err)
39
+
logger.Error("error getting repo", "error", err)
39
40
return echo.NewHTTPError(400, "account not found")
40
41
}
41
42
42
43
if err := bcrypt.CompareHashAndPassword([]byte(urepo.Repo.Password), []byte(req.Password)); err != nil {
43
-
s.logger.Error("password mismatch", "error", err)
44
+
logger.Error("password mismatch", "error", err)
44
45
return echo.NewHTTPError(401, "Invalid did or password")
45
46
}
46
47
47
48
if urepo.Repo.AccountDeleteCode == nil || urepo.Repo.AccountDeleteCodeExpiresAt == nil {
48
-
s.logger.Error("no deletion token found for account")
49
+
logger.Error("no deletion token found for account")
49
50
return echo.NewHTTPError(400, map[string]interface{}{
50
51
"error": "InvalidToken",
51
52
"message": "Token is invalid",
···
53
54
}
54
55
55
56
if *urepo.Repo.AccountDeleteCode != req.Token {
56
-
s.logger.Error("deletion token mismatch")
57
+
logger.Error("deletion token mismatch")
57
58
return echo.NewHTTPError(400, map[string]interface{}{
58
59
"error": "InvalidToken",
59
60
"message": "Token is invalid",
···
61
62
}
62
63
63
64
if time.Now().UTC().After(*urepo.Repo.AccountDeleteCodeExpiresAt) {
64
-
s.logger.Error("deletion token expired")
65
+
logger.Error("deletion token expired")
65
66
return echo.NewHTTPError(400, map[string]interface{}{
66
67
"error": "ExpiredToken",
67
68
"message": "Token is expired",
···
70
71
71
72
tx := s.db.BeginDangerously(ctx)
72
73
if tx.Error != nil {
73
-
s.logger.Error("error starting transaction", "error", tx.Error)
74
+
logger.Error("error starting transaction", "error", tx.Error)
74
75
return helpers.ServerError(e, nil)
75
76
}
76
77
78
+
status := "error"
79
+
func() {
80
+
if status == "error" {
81
+
if err := tx.Rollback().Error; err != nil {
82
+
logger.Error("error rolling back after delete failure", "err", err)
83
+
}
84
+
}
85
+
}()
86
+
77
87
if err := tx.Exec("DELETE FROM blocks WHERE did = ?", nil, req.Did).Error; err != nil {
78
-
tx.Rollback()
79
-
s.logger.Error("error deleting blocks", "error", err)
88
+
logger.Error("error deleting blocks", "error", err)
80
89
return helpers.ServerError(e, nil)
81
90
}
82
91
83
92
if err := tx.Exec("DELETE FROM records WHERE did = ?", nil, req.Did).Error; err != nil {
84
-
tx.Rollback()
85
-
s.logger.Error("error deleting records", "error", err)
93
+
logger.Error("error deleting records", "error", err)
86
94
return helpers.ServerError(e, nil)
87
95
}
88
96
89
97
if err := tx.Exec("DELETE FROM blobs WHERE did = ?", nil, req.Did).Error; err != nil {
90
-
tx.Rollback()
91
-
s.logger.Error("error deleting blobs", "error", err)
98
+
logger.Error("error deleting blobs", "error", err)
92
99
return helpers.ServerError(e, nil)
93
100
}
94
101
95
102
if err := tx.Exec("DELETE FROM tokens WHERE did = ?", nil, req.Did).Error; err != nil {
96
-
tx.Rollback()
97
-
s.logger.Error("error deleting tokens", "error", err)
103
+
logger.Error("error deleting tokens", "error", err)
98
104
return helpers.ServerError(e, nil)
99
105
}
100
106
101
107
if err := tx.Exec("DELETE FROM refresh_tokens WHERE did = ?", nil, req.Did).Error; err != nil {
102
-
tx.Rollback()
103
-
s.logger.Error("error deleting refresh tokens", "error", err)
108
+
logger.Error("error deleting refresh tokens", "error", err)
104
109
return helpers.ServerError(e, nil)
105
110
}
106
111
107
112
if err := tx.Exec("DELETE FROM reserved_keys WHERE did = ?", nil, req.Did).Error; err != nil {
108
-
tx.Rollback()
109
-
s.logger.Error("error deleting reserved keys", "error", err)
113
+
logger.Error("error deleting reserved keys", "error", err)
110
114
return helpers.ServerError(e, nil)
111
115
}
112
116
113
117
if err := tx.Exec("DELETE FROM invite_codes WHERE did = ?", nil, req.Did).Error; err != nil {
114
-
tx.Rollback()
115
-
s.logger.Error("error deleting invite codes", "error", err)
118
+
logger.Error("error deleting invite codes", "error", err)
116
119
return helpers.ServerError(e, nil)
117
120
}
118
121
119
122
if err := tx.Exec("DELETE FROM actors WHERE did = ?", nil, req.Did).Error; err != nil {
120
-
tx.Rollback()
121
-
s.logger.Error("error deleting actor", "error", err)
123
+
logger.Error("error deleting actor", "error", err)
122
124
return helpers.ServerError(e, nil)
123
125
}
124
126
125
127
if err := tx.Exec("DELETE FROM repos WHERE did = ?", nil, req.Did).Error; err != nil {
126
-
tx.Rollback()
127
-
s.logger.Error("error deleting repo", "error", err)
128
+
logger.Error("error deleting repo", "error", err)
128
129
return helpers.ServerError(e, nil)
129
130
}
130
131
132
+
status = "ok"
133
+
131
134
if err := tx.Commit().Error; err != nil {
132
-
s.logger.Error("error committing transaction", "error", err)
135
+
logger.Error("error committing transaction", "error", err)
133
136
return helpers.ServerError(e, nil)
134
137
}
135
138
+7
-5
server/handle_server_get_service_auth.go
+7
-5
server/handle_server_get_service_auth.go
···
25
25
}
26
26
27
27
func (s *Server) handleServerGetServiceAuth(e echo.Context) error {
28
+
logger := s.logger.With("name", "handleServerGetServiceAuth")
29
+
28
30
var req ServerGetServiceAuthRequest
29
31
if err := e.Bind(&req); err != nil {
30
-
s.logger.Error("could not bind service auth request", "error", err)
32
+
logger.Error("could not bind service auth request", "error", err)
31
33
return helpers.ServerError(e, nil)
32
34
}
33
35
···
64
66
}
65
67
hj, err := json.Marshal(header)
66
68
if err != nil {
67
-
s.logger.Error("error marshaling header", "error", err)
69
+
logger.Error("error marshaling header", "error", err)
68
70
return helpers.ServerError(e, nil)
69
71
}
70
72
···
82
84
}
83
85
pj, err := json.Marshal(payload)
84
86
if err != nil {
85
-
s.logger.Error("error marashaling payload", "error", err)
87
+
logger.Error("error marashaling payload", "error", err)
86
88
return helpers.ServerError(e, nil)
87
89
}
88
90
···
93
95
94
96
sk, err := secp256k1secec.NewPrivateKey(repo.SigningKey)
95
97
if err != nil {
96
-
s.logger.Error("can't load private key", "error", err)
98
+
logger.Error("can't load private key", "error", err)
97
99
return err
98
100
}
99
101
100
102
R, S, _, err := sk.SignRaw(rand.Reader, hash[:])
101
103
if err != nil {
102
-
s.logger.Error("error signing", "error", err)
104
+
logger.Error("error signing", "error", err)
103
105
return helpers.ServerError(e, nil)
104
106
}
105
107
+4
-3
server/handle_server_refresh_session.go
+4
-3
server/handle_server_refresh_session.go
···
17
17
18
18
func (s *Server) handleRefreshSession(e echo.Context) error {
19
19
ctx := e.Request().Context()
20
+
logger := s.logger.With("name", "handleServerRefreshSession")
20
21
21
22
token := e.Get("token").(string)
22
23
repo := e.Get("repo").(*models.RepoActor)
23
24
24
25
if err := s.db.Exec(ctx, "DELETE FROM refresh_tokens WHERE token = ?", nil, token).Error; err != nil {
25
-
s.logger.Error("error getting refresh token from db", "error", err)
26
+
logger.Error("error getting refresh token from db", "error", err)
26
27
return helpers.ServerError(e, nil)
27
28
}
28
29
29
30
if err := s.db.Exec(ctx, "DELETE FROM tokens WHERE refresh_token = ?", nil, token).Error; err != nil {
30
-
s.logger.Error("error deleting access token from db", "error", err)
31
+
logger.Error("error deleting access token from db", "error", err)
31
32
return helpers.ServerError(e, nil)
32
33
}
33
34
34
35
sess, err := s.createSession(ctx, &repo.Repo)
35
36
if err != nil {
36
-
s.logger.Error("error creating new session for refresh", "error", err)
37
+
logger.Error("error creating new session for refresh", "error", err)
37
38
return helpers.ServerError(e, nil)
38
39
}
39
40
+3
-2
server/handle_server_request_account_delete.go
+3
-2
server/handle_server_request_account_delete.go
···
11
11
12
12
func (s *Server) handleServerRequestAccountDelete(e echo.Context) error {
13
13
ctx := e.Request().Context()
14
+
logger := s.logger.With("name", "handleServerRequestAccountDelete")
14
15
15
16
urepo := e.Get("repo").(*models.RepoActor)
16
17
···
18
19
expiresAt := time.Now().UTC().Add(15 * time.Minute)
19
20
20
21
if err := s.db.Exec(ctx, "UPDATE repos SET account_delete_code = ?, account_delete_code_expires_at = ? WHERE did = ?", nil, token, expiresAt, urepo.Repo.Did).Error; err != nil {
21
-
s.logger.Error("error setting deletion token", "error", err)
22
+
logger.Error("error setting deletion token", "error", err)
22
23
return helpers.ServerError(e, nil)
23
24
}
24
25
25
26
if urepo.Email != "" {
26
27
if err := s.sendAccountDeleteEmail(urepo.Email, urepo.Actor.Handle, token); err != nil {
27
-
s.logger.Error("error sending account deletion email", "error", err)
28
+
logger.Error("error sending account deletion email", "error", err)
28
29
}
29
30
}
30
31
+3
-2
server/handle_server_request_email_confirmation.go
+3
-2
server/handle_server_request_email_confirmation.go
···
12
12
13
13
func (s *Server) handleServerRequestEmailConfirmation(e echo.Context) error {
14
14
ctx := e.Request().Context()
15
+
logger := s.logger.With("name", "handleServerRequestEmailConfirm")
15
16
16
17
urepo := e.Get("repo").(*models.RepoActor)
17
18
···
23
24
eat := time.Now().Add(10 * time.Minute).UTC()
24
25
25
26
if err := s.db.Exec(ctx, "UPDATE repos SET email_verification_code = ?, email_verification_code_expires_at = ? WHERE did = ?", nil, code, eat, urepo.Repo.Did).Error; err != nil {
26
-
s.logger.Error("error updating user", "error", err)
27
+
logger.Error("error updating user", "error", err)
27
28
return helpers.ServerError(e, nil)
28
29
}
29
30
30
31
if err := s.sendEmailVerification(urepo.Email, urepo.Handle, code); err != nil {
31
-
s.logger.Error("error sending mail", "error", err)
32
+
logger.Error("error sending mail", "error", err)
32
33
return helpers.ServerError(e, nil)
33
34
}
34
35
+3
-2
server/handle_server_request_email_update.go
+3
-2
server/handle_server_request_email_update.go
···
15
15
16
16
func (s *Server) handleServerRequestEmailUpdate(e echo.Context) error {
17
17
ctx := e.Request().Context()
18
+
logger := s.logger.With("name", "handleServerRequestEmailUpdate")
18
19
19
20
urepo := e.Get("repo").(*models.RepoActor)
20
21
···
23
24
eat := time.Now().Add(10 * time.Minute).UTC()
24
25
25
26
if err := s.db.Exec(ctx, "UPDATE repos SET email_update_code = ?, email_update_code_expires_at = ? WHERE did = ?", nil, code, eat, urepo.Repo.Did).Error; err != nil {
26
-
s.logger.Error("error updating repo", "error", err)
27
+
logger.Error("error updating repo", "error", err)
27
28
return helpers.ServerError(e, nil)
28
29
}
29
30
30
31
if err := s.sendEmailUpdate(urepo.Email, urepo.Handle, code); err != nil {
31
-
s.logger.Error("error sending email", "error", err)
32
+
logger.Error("error sending email", "error", err)
32
33
return helpers.ServerError(e, nil)
33
34
}
34
35
}
+3
-2
server/handle_server_request_password_reset.go
+3
-2
server/handle_server_request_password_reset.go
···
15
15
16
16
func (s *Server) handleServerRequestPasswordReset(e echo.Context) error {
17
17
ctx := e.Request().Context()
18
+
logger := s.logger.With("name", "handleServerRequestPasswordReset")
18
19
19
20
urepo, ok := e.Get("repo").(*models.RepoActor)
20
21
if !ok {
···
39
40
eat := time.Now().Add(10 * time.Minute).UTC()
40
41
41
42
if err := s.db.Exec(ctx, "UPDATE repos SET password_reset_code = ?, password_reset_code_expires_at = ? WHERE did = ?", nil, code, eat, urepo.Repo.Did).Error; err != nil {
42
-
s.logger.Error("error updating repo", "error", err)
43
+
logger.Error("error updating repo", "error", err)
43
44
return helpers.ServerError(e, nil)
44
45
}
45
46
46
47
if err := s.sendPasswordReset(urepo.Email, urepo.Handle, code); err != nil {
47
-
s.logger.Error("error sending email", "error", err)
48
+
logger.Error("error sending email", "error", err)
48
49
return helpers.ServerError(e, nil)
49
50
}
50
51
+6
-5
server/handle_server_reserve_signing_key.go
+6
-5
server/handle_server_reserve_signing_key.go
···
20
20
21
21
func (s *Server) handleServerReserveSigningKey(e echo.Context) error {
22
22
ctx := e.Request().Context()
23
+
logger := s.logger.With("name", "handleServerReserveSigningKey")
23
24
24
25
var req ServerReserveSigningKeyRequest
25
26
if err := e.Bind(&req); err != nil {
26
-
s.logger.Error("could not bind reserve signing key request", "error", err)
27
+
logger.Error("could not bind reserve signing key request", "error", err)
27
28
return helpers.ServerError(e, nil)
28
29
}
29
30
···
38
39
39
40
k, err := atcrypto.GeneratePrivateKeyK256()
40
41
if err != nil {
41
-
s.logger.Error("error creating signing key", "endpoint", "com.atproto.server.reserveSigningKey", "error", err)
42
+
logger.Error("error creating signing key", "endpoint", "com.atproto.server.reserveSigningKey", "error", err)
42
43
return helpers.ServerError(e, nil)
43
44
}
44
45
45
46
pubKey, err := k.PublicKey()
46
47
if err != nil {
47
-
s.logger.Error("error getting public key", "endpoint", "com.atproto.server.reserveSigningKey", "error", err)
48
+
logger.Error("error getting public key", "endpoint", "com.atproto.server.reserveSigningKey", "error", err)
48
49
return helpers.ServerError(e, nil)
49
50
}
50
51
···
58
59
}
59
60
60
61
if err := s.db.Create(ctx, &reservedKey, nil).Error; err != nil {
61
-
s.logger.Error("error storing reserved key", "endpoint", "com.atproto.server.reserveSigningKey", "error", err)
62
+
logger.Error("error storing reserved key", "endpoint", "com.atproto.server.reserveSigningKey", "error", err)
62
63
return helpers.ServerError(e, nil)
63
64
}
64
65
65
-
s.logger.Info("reserved signing key", "keyDid", keyDid, "forDid", req.Did)
66
+
logger.Info("reserved signing key", "keyDid", keyDid, "forDid", req.Did)
66
67
67
68
return e.JSON(200, ServerReserveSigningKeyResponse{
68
69
SigningKey: keyDid,
+4
-3
server/handle_server_reset_password.go
+4
-3
server/handle_server_reset_password.go
···
17
17
18
18
func (s *Server) handleServerResetPassword(e echo.Context) error {
19
19
ctx := e.Request().Context()
20
+
logger := s.logger.With("name", "handleServerResetPassword")
20
21
21
22
urepo := e.Get("repo").(*models.RepoActor)
22
23
23
24
var req ComAtprotoServerResetPasswordRequest
24
25
if err := e.Bind(&req); err != nil {
25
-
s.logger.Error("error binding", "error", err)
26
+
logger.Error("error binding", "error", err)
26
27
return helpers.ServerError(e, nil)
27
28
}
28
29
···
44
45
45
46
hash, err := bcrypt.GenerateFromPassword([]byte(req.Password), 10)
46
47
if err != nil {
47
-
s.logger.Error("error creating hash", "error", err)
48
+
logger.Error("error creating hash", "error", err)
48
49
return helpers.ServerError(e, nil)
49
50
}
50
51
51
52
if err := s.db.Exec(ctx, "UPDATE repos SET password_reset_code = NULL, password_reset_code_expires_at = NULL, password = ? WHERE did = ?", nil, hash, urepo.Repo.Did).Error; err != nil {
52
-
s.logger.Error("error updating repo", "error", err)
53
+
logger.Error("error updating repo", "error", err)
53
54
return helpers.ServerError(e, nil)
54
55
}
55
56
+3
-1
server/handle_server_resolve_handle.go
+3
-1
server/handle_server_resolve_handle.go
···
10
10
)
11
11
12
12
func (s *Server) handleResolveHandle(e echo.Context) error {
13
+
logger := s.logger.With("name", "handleServerResolveHandle")
14
+
13
15
type Resp struct {
14
16
Did string `json:"did"`
15
17
}
···
28
30
ctx := context.WithValue(e.Request().Context(), "skip-cache", true)
29
31
did, err := s.passport.ResolveHandle(ctx, parsed.String())
30
32
if err != nil {
31
-
s.logger.Error("error resolving handle", "error", err)
33
+
logger.Error("error resolving handle", "error", err)
32
34
return helpers.ServerError(e, nil)
33
35
}
34
36
+3
-2
server/handle_server_update_email.go
+3
-2
server/handle_server_update_email.go
···
16
16
17
17
func (s *Server) handleServerUpdateEmail(e echo.Context) error {
18
18
ctx := e.Request().Context()
19
+
logger := s.logger.With("name", "handleServerUpdateEmail")
19
20
20
21
urepo := e.Get("repo").(*models.RepoActor)
21
22
22
23
var req ComAtprotoServerUpdateEmailRequest
23
24
if err := e.Bind(&req); err != nil {
24
-
s.logger.Error("error binding", "error", err)
25
+
logger.Error("error binding", "error", err)
25
26
return helpers.ServerError(e, nil)
26
27
}
27
28
···
42
43
}
43
44
44
45
if err := s.db.Exec(ctx, "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 {
45
-
s.logger.Error("error updating repo", "error", err)
46
+
logger.Error("error updating repo", "error", err)
46
47
return helpers.ServerError(e, nil)
47
48
}
48
49
+9
-8
server/handle_sync_get_blob.go
+9
-8
server/handle_sync_get_blob.go
···
18
18
19
19
func (s *Server) handleSyncGetBlob(e echo.Context) error {
20
20
ctx := e.Request().Context()
21
+
logger := s.logger.With("name", "handleSyncGetBlob")
21
22
22
23
did := e.QueryParam("did")
23
24
if did == "" {
···
36
37
37
38
urepo, err := s.getRepoActorByDid(ctx, did)
38
39
if err != nil {
39
-
s.logger.Error("could not find user for requested blob", "error", err)
40
+
logger.Error("could not find user for requested blob", "error", err)
40
41
return helpers.InputError(e, nil)
41
42
}
42
43
···
49
50
50
51
var blob models.Blob
51
52
if err := s.db.Raw(ctx, "SELECT * FROM blobs WHERE did = ? AND cid = ?", nil, did, c.Bytes()).Scan(&blob).Error; err != nil {
52
-
s.logger.Error("error looking up blob", "error", err)
53
+
logger.Error("error looking up blob", "error", err)
53
54
return helpers.ServerError(e, nil)
54
55
}
55
56
···
58
59
if blob.Storage == "sqlite" {
59
60
var parts []models.BlobPart
60
61
if err := s.db.Raw(ctx, "SELECT * FROM blob_parts WHERE blob_id = ? ORDER BY idx", nil, blob.ID).Scan(&parts).Error; err != nil {
61
-
s.logger.Error("error getting blob parts", "error", err)
62
+
logger.Error("error getting blob parts", "error", err)
62
63
return helpers.ServerError(e, nil)
63
64
}
64
65
···
68
69
}
69
70
} else if blob.Storage == "s3" {
70
71
if !(s.s3Config != nil && s.s3Config.BlobstoreEnabled) {
71
-
s.logger.Error("s3 storage disabled")
72
+
logger.Error("s3 storage disabled")
72
73
return helpers.ServerError(e, nil)
73
74
}
74
75
···
91
92
92
93
sess, err := session.NewSession(config)
93
94
if err != nil {
94
-
s.logger.Error("error creating aws session", "error", err)
95
+
logger.Error("error creating aws session", "error", err)
95
96
return helpers.ServerError(e, nil)
96
97
}
97
98
···
100
101
Bucket: aws.String(s.s3Config.Bucket),
101
102
Key: aws.String(blobKey),
102
103
}); err != nil {
103
-
s.logger.Error("error getting blob from s3", "error", err)
104
+
logger.Error("error getting blob from s3", "error", err)
104
105
return helpers.ServerError(e, nil)
105
106
} else {
106
107
read := 0
···
114
115
break
115
116
}
116
117
} else if err != nil && err != io.ErrUnexpectedEOF {
117
-
s.logger.Error("error reading blob", "error", err)
118
+
logger.Error("error reading blob", "error", err)
118
119
return helpers.ServerError(e, nil)
119
120
}
120
121
···
125
126
}
126
127
}
127
128
} else {
128
-
s.logger.Error("unknown storage", "storage", blob.Storage)
129
+
logger.Error("unknown storage", "storage", blob.Storage)
129
130
return helpers.ServerError(e, nil)
130
131
}
131
132
+2
-1
server/handle_sync_get_blocks.go
+2
-1
server/handle_sync_get_blocks.go
···
18
18
19
19
func (s *Server) handleGetBlocks(e echo.Context) error {
20
20
ctx := e.Request().Context()
21
+
logger := s.logger.With("name", "handleSyncGetBlocks")
21
22
22
23
var req ComAtprotoSyncGetBlocksRequest
23
24
if err := e.Bind(&req); err != nil {
···
52
53
})
53
54
54
55
if _, err := carstore.LdWrite(buf, hb); err != nil {
55
-
s.logger.Error("error writing to car", "error", err)
56
+
logger.Error("error writing to car", "error", err)
56
57
return helpers.ServerError(e, nil)
57
58
}
58
59
+4
-3
server/handle_sync_get_record.go
+4
-3
server/handle_sync_get_record.go
···
14
14
15
15
func (s *Server) handleSyncGetRecord(e echo.Context) error {
16
16
ctx := e.Request().Context()
17
+
logger := s.logger.With("name", "handleSyncGetRecord")
17
18
18
19
did := e.QueryParam("did")
19
20
collection := e.QueryParam("collection")
···
21
22
22
23
var urepo models.Repo
23
24
if err := s.db.Raw(ctx, "SELECT * FROM repos WHERE did = ?", nil, did).Scan(&urepo).Error; err != nil {
24
-
s.logger.Error("error getting repo", "error", err)
25
+
logger.Error("error getting repo", "error", err)
25
26
return helpers.ServerError(e, nil)
26
27
}
27
28
···
38
39
})
39
40
40
41
if _, err := carstore.LdWrite(buf, hb); err != nil {
41
-
s.logger.Error("error writing to car", "error", err)
42
+
logger.Error("error writing to car", "error", err)
42
43
return helpers.ServerError(e, nil)
43
44
}
44
45
45
46
for _, blk := range blocks {
46
47
if _, err := carstore.LdWrite(buf, blk.Cid().Bytes(), blk.RawData()); err != nil {
47
-
s.logger.Error("error writing to car", "error", err)
48
+
logger.Error("error writing to car", "error", err)
48
49
return helpers.ServerError(e, nil)
49
50
}
50
51
}
+2
-1
server/handle_sync_get_repo.go
+2
-1
server/handle_sync_get_repo.go
···
14
14
15
15
func (s *Server) handleSyncGetRepo(e echo.Context) error {
16
16
ctx := e.Request().Context()
17
+
logger := s.logger.With("name", "handleSyncGetRepo")
17
18
18
19
did := e.QueryParam("did")
19
20
if did == "" {
···
38
39
buf := new(bytes.Buffer)
39
40
40
41
if _, err := carstore.LdWrite(buf, hb); err != nil {
41
-
s.logger.Error("error writing to car", "error", err)
42
+
logger.Error("error writing to car", "error", err)
42
43
return helpers.ServerError(e, nil)
43
44
}
44
45
+4
-3
server/handle_sync_list_blobs.go
+4
-3
server/handle_sync_list_blobs.go
···
15
15
16
16
func (s *Server) handleSyncListBlobs(e echo.Context) error {
17
17
ctx := e.Request().Context()
18
+
logger := s.logger.With("name", "handleSyncListBlobs")
18
19
19
20
did := e.QueryParam("did")
20
21
if did == "" {
···
39
40
40
41
urepo, err := s.getRepoActorByDid(ctx, did)
41
42
if err != nil {
42
-
s.logger.Error("could not find user for requested blobs", "error", err)
43
+
logger.Error("could not find user for requested blobs", "error", err)
43
44
return helpers.InputError(e, nil)
44
45
}
45
46
···
52
53
53
54
var blobs []models.Blob
54
55
if err := s.db.Raw(ctx, "SELECT * FROM blobs WHERE did = ? "+cursorquery+" ORDER BY created_at DESC LIMIT ?", nil, params...).Scan(&blobs).Error; err != nil {
55
-
s.logger.Error("error getting records", "error", err)
56
+
logger.Error("error getting records", "error", err)
56
57
return helpers.ServerError(e, nil)
57
58
}
58
59
···
60
61
for _, b := range blobs {
61
62
c, err := cid.Cast(b.Cid)
62
63
if err != nil {
63
-
s.logger.Error("error casting cid", "error", err)
64
+
logger.Error("error casting cid", "error", err)
64
65
return helpers.ServerError(e, nil)
65
66
}
66
67
cstrs = append(cstrs, c.String())
+54
-42
server/handle_sync_subscribe_repos.go
+54
-42
server/handle_sync_subscribe_repos.go
···
7
7
"github.com/bluesky-social/indigo/events"
8
8
"github.com/bluesky-social/indigo/lex/util"
9
9
"github.com/btcsuite/websocket"
10
+
"github.com/haileyok/cocoon/metrics"
10
11
"github.com/labstack/echo/v4"
11
12
)
12
13
···
24
25
logger = logger.With("ident", ident)
25
26
logger.Info("new connection established")
26
27
28
+
metrics.RelaysConnected.WithLabelValues(ident).Inc()
29
+
defer func() {
30
+
metrics.RelaysConnected.WithLabelValues(ident).Dec()
31
+
}()
32
+
27
33
evts, cancel, err := s.evtman.Subscribe(ctx, ident, func(evt *events.XRPCStreamEvent) bool {
28
34
return true
29
35
}, nil)
···
34
40
35
41
header := events.EventHeader{Op: events.EvtKindMessage}
36
42
for evt := range evts {
37
-
wc, err := conn.NextWriter(websocket.BinaryMessage)
38
-
if err != nil {
39
-
logger.Error("error writing message to relay", "err", err)
40
-
break
41
-
}
43
+
func() {
44
+
defer func() {
45
+
metrics.RelaySends.WithLabelValues(header.MsgType).Inc()
46
+
}()
42
47
43
-
if ctx.Err() != nil {
44
-
logger.Error("context error", "err", err)
45
-
break
46
-
}
48
+
wc, err := conn.NextWriter(websocket.BinaryMessage)
49
+
if err != nil {
50
+
logger.Error("error writing message to relay", "err", err)
51
+
return
52
+
}
47
53
48
-
var obj util.CBOR
49
-
switch {
50
-
case evt.Error != nil:
51
-
header.Op = events.EvtKindErrorFrame
52
-
obj = evt.Error
53
-
case evt.RepoCommit != nil:
54
-
header.MsgType = "#commit"
55
-
obj = evt.RepoCommit
56
-
case evt.RepoIdentity != nil:
57
-
header.MsgType = "#identity"
58
-
obj = evt.RepoIdentity
59
-
case evt.RepoAccount != nil:
60
-
header.MsgType = "#account"
61
-
obj = evt.RepoAccount
62
-
case evt.RepoInfo != nil:
63
-
header.MsgType = "#info"
64
-
obj = evt.RepoInfo
65
-
default:
66
-
logger.Warn("unrecognized event kind")
67
-
return nil
68
-
}
54
+
if ctx.Err() != nil {
55
+
logger.Error("context error", "err", err)
56
+
return
57
+
}
69
58
70
-
if err := header.MarshalCBOR(wc); err != nil {
71
-
logger.Error("failed to write header to relay", "err", err)
72
-
break
73
-
}
59
+
var obj util.CBOR
60
+
switch {
61
+
case evt.Error != nil:
62
+
header.Op = events.EvtKindErrorFrame
63
+
obj = evt.Error
64
+
case evt.RepoCommit != nil:
65
+
header.MsgType = "#commit"
66
+
obj = evt.RepoCommit
67
+
case evt.RepoIdentity != nil:
68
+
header.MsgType = "#identity"
69
+
obj = evt.RepoIdentity
70
+
case evt.RepoAccount != nil:
71
+
header.MsgType = "#account"
72
+
obj = evt.RepoAccount
73
+
case evt.RepoInfo != nil:
74
+
header.MsgType = "#info"
75
+
obj = evt.RepoInfo
76
+
default:
77
+
logger.Warn("unrecognized event kind")
78
+
return
79
+
}
74
80
75
-
if err := obj.MarshalCBOR(wc); err != nil {
76
-
logger.Error("failed to write event to relay", "err", err)
77
-
break
78
-
}
81
+
if err := header.MarshalCBOR(wc); err != nil {
82
+
logger.Error("failed to write header to relay", "err", err)
83
+
return
84
+
}
79
85
80
-
if err := wc.Close(); err != nil {
81
-
logger.Error("failed to flush-close our event write", "err", err)
82
-
break
83
-
}
86
+
if err := obj.MarshalCBOR(wc); err != nil {
87
+
logger.Error("failed to write event to relay", "err", err)
88
+
return
89
+
}
90
+
91
+
if err := wc.Close(); err != nil {
92
+
logger.Error("failed to flush-close our event write", "err", err)
93
+
return
94
+
}
95
+
}()
84
96
}
85
97
86
98
// we should tell the relay to request a new crawl at this point if we got disconnected
+2
-1
server/handle_well_known.go
+2
-1
server/handle_well_known.go
···
68
68
69
69
func (s *Server) handleAtprotoDid(e echo.Context) error {
70
70
ctx := e.Request().Context()
71
+
logger := s.logger.With("name", "handleAtprotoDid")
71
72
72
73
host := e.Request().Host
73
74
if host == "" {
···
91
92
if err == gorm.ErrRecordNotFound {
92
93
return e.NoContent(404)
93
94
}
94
-
s.logger.Error("error looking up actor by handle", "error", err)
95
+
logger.Error("error looking up actor by handle", "error", err)
95
96
return helpers.ServerError(e, nil)
96
97
}
97
98
+18
-16
server/middleware.go
+18
-16
server/middleware.go
···
38
38
func (s *Server) handleLegacySessionMiddleware(next echo.HandlerFunc) echo.HandlerFunc {
39
39
return func(e echo.Context) error {
40
40
ctx := e.Request().Context()
41
+
logger := s.logger.With("name", "handleLegacySessionMiddleware")
41
42
42
43
authheader := e.Request().Header.Get("authorization")
43
44
if authheader == "" {
···
69
70
if hasLxm {
70
71
pts := strings.Split(e.Request().URL.String(), "/")
71
72
if lxm != pts[len(pts)-1] {
72
-
s.logger.Error("service auth lxm incorrect", "lxm", lxm, "expected", pts[len(pts)-1], "error", err)
73
+
logger.Error("service auth lxm incorrect", "lxm", lxm, "expected", pts[len(pts)-1], "error", err)
73
74
return helpers.InputError(e, nil)
74
75
}
75
76
76
77
maybeDid, ok := claims["iss"].(string)
77
78
if !ok {
78
-
s.logger.Error("no iss in service auth token", "error", err)
79
+
logger.Error("no iss in service auth token", "error", err)
79
80
return helpers.InputError(e, nil)
80
81
}
81
82
did = maybeDid
82
83
83
84
maybeRepo, err := s.getRepoActorByDid(ctx, did)
84
85
if err != nil {
85
-
s.logger.Error("error fetching repo", "error", err)
86
+
logger.Error("error fetching repo", "error", err)
86
87
return helpers.ServerError(e, nil)
87
88
}
88
89
repo = maybeRepo
···
96
97
return s.privateKey.Public(), nil
97
98
})
98
99
if err != nil {
99
-
s.logger.Error("error parsing jwt", "error", err)
100
+
logger.Error("error parsing jwt", "error", err)
100
101
return helpers.ExpiredTokenError(e)
101
102
}
102
103
···
109
110
hash := sha256.Sum256([]byte(signingInput))
110
111
sigBytes, err := base64.RawURLEncoding.DecodeString(kpts[2])
111
112
if err != nil {
112
-
s.logger.Error("error decoding signature bytes", "error", err)
113
+
logger.Error("error decoding signature bytes", "error", err)
113
114
return helpers.ServerError(e, nil)
114
115
}
115
116
116
117
if len(sigBytes) != 64 {
117
-
s.logger.Error("incorrect sigbytes length", "length", len(sigBytes))
118
+
logger.Error("incorrect sigbytes length", "length", len(sigBytes))
118
119
return helpers.ServerError(e, nil)
119
120
}
120
121
···
140
141
141
142
sk, err := secp256k1secec.NewPrivateKey(repo.SigningKey)
142
143
if err != nil {
143
-
s.logger.Error("can't load private key", "error", err)
144
+
logger.Error("can't load private key", "error", err)
144
145
return err
145
146
}
146
147
147
148
pubKey, ok := sk.Public().(*secp256k1secec.PublicKey)
148
149
if !ok {
149
-
s.logger.Error("error getting public key from sk")
150
+
logger.Error("error getting public key from sk")
150
151
return helpers.ServerError(e, nil)
151
152
}
152
153
153
154
verified := pubKey.VerifyRaw(hash[:], rr, ss)
154
155
if !verified {
155
-
s.logger.Error("error verifying", "error", err)
156
+
logger.Error("error verifying", "error", err)
156
157
return helpers.ServerError(e, nil)
157
158
}
158
159
}
···
181
182
return helpers.InvalidTokenError(e)
182
183
}
183
184
184
-
s.logger.Error("error getting token from db", "error", err)
185
+
logger.Error("error getting token from db", "error", err)
185
186
return helpers.ServerError(e, nil)
186
187
}
187
188
···
192
193
193
194
exp, ok := claims["exp"].(float64)
194
195
if !ok {
195
-
s.logger.Error("error getting iat from token")
196
+
logger.Error("error getting iat from token")
196
197
return helpers.ServerError(e, nil)
197
198
}
198
199
···
203
204
if repo == nil {
204
205
maybeRepo, err := s.getRepoActorByDid(ctx, claims["sub"].(string))
205
206
if err != nil {
206
-
s.logger.Error("error fetching repo", "error", err)
207
+
logger.Error("error fetching repo", "error", err)
207
208
return helpers.ServerError(e, nil)
208
209
}
209
210
repo = maybeRepo
···
225
226
func (s *Server) handleOauthSessionMiddleware(next echo.HandlerFunc) echo.HandlerFunc {
226
227
return func(e echo.Context) error {
227
228
ctx := e.Request().Context()
229
+
logger := s.logger.With("name", "handleOauthSessionMiddleware")
228
230
229
231
authheader := e.Request().Header.Get("authorization")
230
232
if authheader == "" {
···
257
259
"error": "use_dpop_nonce",
258
260
})
259
261
}
260
-
s.logger.Error("invalid dpop proof", "error", err)
262
+
logger.Error("invalid dpop proof", "error", err)
261
263
return helpers.InputError(e, nil)
262
264
}
263
265
264
266
var oauthToken provider.OauthToken
265
267
if err := s.db.Raw(ctx, "SELECT * FROM oauth_tokens WHERE token = ?", nil, accessToken).Scan(&oauthToken).Error; err != nil {
266
-
s.logger.Error("error finding access token in db", "error", err)
268
+
logger.Error("error finding access token in db", "error", err)
267
269
return helpers.InputError(e, nil)
268
270
}
269
271
···
272
274
}
273
275
274
276
if *oauthToken.Parameters.DpopJkt != proof.JKT {
275
-
s.logger.Error("jkt mismatch", "token", oauthToken.Parameters.DpopJkt, "proof", proof.JKT)
277
+
logger.Error("jkt mismatch", "token", oauthToken.Parameters.DpopJkt, "proof", proof.JKT)
276
278
return helpers.InputError(e, to.StringPtr("dpop jkt mismatch"))
277
279
}
278
280
···
287
289
288
290
repo, err := s.getRepoActorByDid(ctx, oauthToken.Sub)
289
291
if err != nil {
290
-
s.logger.Error("could not find actor in db", "error", err)
292
+
logger.Error("could not find actor in db", "error", err)
291
293
return helpers.ServerError(e, nil)
292
294
}
293
295
+7
server/repo.go
+7
server/repo.go
···
17
17
lexutil "github.com/bluesky-social/indigo/lex/util"
18
18
"github.com/bluesky-social/indigo/repo"
19
19
"github.com/haileyok/cocoon/internal/db"
20
+
"github.com/haileyok/cocoon/metrics"
20
21
"github.com/haileyok/cocoon/models"
21
22
"github.com/haileyok/cocoon/recording_blockstore"
22
23
blocks "github.com/ipfs/go-block-format"
···
249
250
newroot, rev, err := r.Commit(ctx, urepo.SignFor)
250
251
if err != nil {
251
252
return nil, err
253
+
}
254
+
255
+
for _, result := range results {
256
+
if result.Type != nil {
257
+
metrics.RepoOperations.WithLabelValues(*result.Type).Inc()
258
+
}
252
259
}
253
260
254
261
// create a buffer for dumping our new cbor into
+38
-27
server/server.go
+38
-27
server/server.go
···
39
39
"github.com/haileyok/cocoon/oauth/provider"
40
40
"github.com/haileyok/cocoon/plc"
41
41
"github.com/ipfs/go-cid"
42
+
"github.com/labstack/echo-contrib/echoprometheus"
42
43
echo_session "github.com/labstack/echo-contrib/session"
43
44
"github.com/labstack/echo/v4"
44
45
"github.com/labstack/echo/v4/middleware"
···
89
90
}
90
91
91
92
type Args struct {
93
+
Logger *slog.Logger
94
+
92
95
Addr string
93
96
DbName string
94
97
DbType string
95
98
DatabaseURL string
96
-
Logger *slog.Logger
97
99
Version string
98
100
Did string
99
101
Hostname string
···
209
211
}
210
212
211
213
func New(args *Args) (*Server, error) {
214
+
if args.Logger == nil {
215
+
args.Logger = slog.Default()
216
+
}
217
+
218
+
logger := args.Logger.With("name", "New")
219
+
212
220
if args.Addr == "" {
213
221
return nil, fmt.Errorf("addr must be set")
214
222
}
···
237
245
return nil, fmt.Errorf("admin password must be set")
238
246
}
239
247
240
-
if args.Logger == nil {
241
-
args.Logger = slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{}))
242
-
}
243
-
244
248
if args.SessionSecret == "" {
245
249
panic("SESSION SECRET WAS NOT SET. THIS IS REQUIRED. ")
246
250
}
···
248
252
e := echo.New()
249
253
250
254
e.Pre(middleware.RemoveTrailingSlash())
251
-
e.Pre(slogecho.New(args.Logger))
255
+
e.Pre(slogecho.New(args.Logger.With("component", "slogecho")))
252
256
e.Use(echo_session.Middleware(sessions.NewCookieStore([]byte(args.SessionSecret))))
257
+
e.Use(echoprometheus.NewMiddleware("cocoon"))
253
258
e.Use(middleware.CORSWithConfig(middleware.CORSConfig{
254
259
AllowOrigins: []string{"*"},
255
260
AllowHeaders: []string{"*"},
···
311
316
if err != nil {
312
317
return nil, fmt.Errorf("failed to connect to postgres: %w", err)
313
318
}
314
-
args.Logger.Info("connected to PostgreSQL database")
319
+
logger.Info("connected to PostgreSQL database")
315
320
default:
316
321
gdb, err = gorm.Open(sqlite.Open(args.DbName), &gorm.Config{})
317
322
if err != nil {
318
323
return nil, fmt.Errorf("failed to open sqlite database: %w", err)
319
324
}
320
-
args.Logger.Info("connected to SQLite database", "path", args.DbName)
325
+
logger.Info("connected to SQLite database", "path", args.DbName)
321
326
}
322
327
dbw := db.NewDB(gdb)
323
328
···
360
365
var nonceSecret []byte
361
366
maybeSecret, err := os.ReadFile("nonce.secret")
362
367
if err != nil && !os.IsNotExist(err) {
363
-
args.Logger.Error("error attempting to read nonce secret", "error", err)
368
+
logger.Error("error attempting to read nonce secret", "error", err)
364
369
} else {
365
370
nonceSecret = maybeSecret
366
371
}
···
398
403
Hostname: args.Hostname,
399
404
ClientManagerArgs: client.ManagerArgs{
400
405
Cli: oauthCli,
401
-
Logger: args.Logger,
406
+
Logger: args.Logger.With("component", "oauth-client-manager"),
402
407
},
403
408
DpopManagerArgs: dpop.ManagerArgs{
404
409
NonceSecret: nonceSecret,
405
410
NonceRotationInterval: constants.NonceMaxRotationInterval / 3,
406
411
OnNonceSecretCreated: func(newNonce []byte) {
407
412
if err := os.WriteFile("nonce.secret", newNonce, 0644); err != nil {
408
-
args.Logger.Error("error writing new nonce secret", "error", err)
413
+
logger.Error("error writing new nonce secret", "error", err)
409
414
}
410
415
},
411
-
Logger: args.Logger,
416
+
Logger: args.Logger.With("component", "dpop-manager"),
412
417
Hostname: args.Hostname,
413
418
},
414
419
}),
···
535
540
}
536
541
537
542
func (s *Server) Serve(ctx context.Context) error {
543
+
logger := s.logger.With("name", "Serve")
544
+
538
545
s.addRoutes()
539
546
540
-
s.logger.Info("migrating...")
547
+
logger.Info("migrating...")
541
548
542
549
s.db.AutoMigrate(
543
550
&models.Actor{},
···
554
561
&provider.OauthAuthorizationRequest{},
555
562
)
556
563
557
-
s.logger.Info("starting cocoon")
564
+
logger.Info("starting cocoon")
558
565
559
566
go func() {
560
567
if err := s.httpd.ListenAndServe(); err != nil {
···
566
573
567
574
go func() {
568
575
if err := s.requestCrawl(ctx); err != nil {
569
-
s.logger.Error("error requesting crawls", "err", err)
576
+
logger.Error("error requesting crawls", "err", err)
570
577
}
571
578
}()
572
579
···
584
591
585
592
logger.Info("requesting crawl with configured relays")
586
593
587
-
if time.Now().Sub(s.lastRequestCrawl) <= 1*time.Minute {
594
+
if time.Since(s.lastRequestCrawl) <= 1*time.Minute {
588
595
return fmt.Errorf("a crawl request has already been made within the last minute")
589
596
}
590
597
···
607
614
}
608
615
609
616
func (s *Server) doBackup() {
617
+
logger := s.logger.With("name", "doBackup")
618
+
610
619
if s.dbType == "postgres" {
611
-
s.logger.Info("skipping S3 backup - PostgreSQL backups should be handled externally (pg_dump, managed database backups, etc.)")
620
+
logger.Info("skipping S3 backup - PostgreSQL backups should be handled externally (pg_dump, managed database backups, etc.)")
612
621
return
613
622
}
614
623
615
624
start := time.Now()
616
625
617
-
s.logger.Info("beginning backup to s3...")
626
+
logger.Info("beginning backup to s3...")
618
627
619
628
var buf bytes.Buffer
620
629
if err := func() error {
621
-
s.logger.Info("reading database bytes...")
630
+
logger.Info("reading database bytes...")
622
631
s.db.Lock()
623
632
defer s.db.Unlock()
624
633
···
634
643
635
644
return nil
636
645
}(); err != nil {
637
-
s.logger.Error("error backing up database", "error", err)
646
+
logger.Error("error backing up database", "error", err)
638
647
return
639
648
}
640
649
641
650
if err := func() error {
642
-
s.logger.Info("sending to s3...")
651
+
logger.Info("sending to s3...")
643
652
644
653
currTime := time.Now().Format("2006-01-02_15-04-05")
645
654
key := "cocoon-backup-" + currTime + ".db"
···
669
678
return fmt.Errorf("error uploading file to s3: %w", err)
670
679
}
671
680
672
-
s.logger.Info("finished uploading backup to s3", "key", key, "duration", time.Now().Sub(start).Seconds())
681
+
logger.Info("finished uploading backup to s3", "key", key, "duration", time.Now().Sub(start).Seconds())
673
682
674
683
return nil
675
684
}(); err != nil {
676
-
s.logger.Error("error uploading database backup", "error", err)
685
+
logger.Error("error uploading database backup", "error", err)
677
686
return
678
687
}
679
688
···
681
690
}
682
691
683
692
func (s *Server) backupRoutine() {
693
+
logger := s.logger.With("name", "backupRoutine")
694
+
684
695
if s.s3Config == nil || !s.s3Config.BackupsEnabled {
685
696
return
686
697
}
687
698
688
699
if s.s3Config.Region == "" {
689
-
s.logger.Warn("no s3 region configured but backups are enabled. backups will not run.")
700
+
logger.Warn("no s3 region configured but backups are enabled. backups will not run.")
690
701
return
691
702
}
692
703
693
704
if s.s3Config.Bucket == "" {
694
-
s.logger.Warn("no s3 bucket configured but backups are enabled. backups will not run.")
705
+
logger.Warn("no s3 bucket configured but backups are enabled. backups will not run.")
695
706
return
696
707
}
697
708
698
709
if s.s3Config.AccessKey == "" {
699
-
s.logger.Warn("no s3 access key configured but backups are enabled. backups will not run.")
710
+
logger.Warn("no s3 access key configured but backups are enabled. backups will not run.")
700
711
return
701
712
}
702
713
703
714
if s.s3Config.SecretKey == "" {
704
-
s.logger.Warn("no s3 secret key configured but backups are enabled. backups will not run.")
715
+
logger.Warn("no s3 secret key configured but backups are enabled. backups will not run.")
705
716
return
706
717
}
707
718
+1
-1
test.go
+1
-1
test.go
···
32
32
33
33
u.Path = "xrpc/com.atproto.sync.subscribeRepos"
34
34
conn, _, err := dialer.Dial(u.String(), http.Header{
35
-
"User-Agent": []string{fmt.Sprintf("hot-topic/0.0.0")},
35
+
"User-Agent": []string{"cocoon-test/0.0.0"},
36
36
})
37
37
if err != nil {
38
38
return fmt.Errorf("subscribing to firehose failed (dialing): %w", err)