Lasa is a stateless proxy that generates a RSS or an Atom feed from a Standard.site publication.
lasa.anhgelus.world
rss
atom
atprotocol
standard-site
atproto
refactor(server): idiomatic middlewares
+130
-69
+97
cmd/internal/mux.go
+97
cmd/internal/mux.go
···
1
+
package internal
2
+
3
+
import (
4
+
"context"
5
+
"errors"
6
+
"fmt"
7
+
"log/slog"
8
+
"net/http"
9
+
"slices"
10
+
"time"
11
+
)
12
+
13
+
type StatusWriter struct {
14
+
http.ResponseWriter
15
+
Code int
16
+
}
17
+
18
+
func (w *StatusWriter) WriteHeader(statusCode int) {
19
+
w.ResponseWriter.WriteHeader(statusCode)
20
+
w.Code = statusCode
21
+
}
22
+
23
+
type Handler func(*StatusWriter, *http.Request)
24
+
25
+
type Middleware func(Handler, *StatusWriter, *http.Request)
26
+
27
+
type Mux struct {
28
+
middlewares []func(Handler) Handler
29
+
handler Handler
30
+
}
31
+
32
+
func NewMux(base *http.ServeMux) *Mux {
33
+
return &Mux{handler: func(w *StatusWriter, r *http.Request) {
34
+
base.ServeHTTP(w, r)
35
+
}}
36
+
}
37
+
38
+
func (m *Mux) Handle() http.Handler {
39
+
slices.Reverse(m.middlewares)
40
+
for _, middle := range m.middlewares {
41
+
m.handler = middle(m.handler)
42
+
}
43
+
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
44
+
m.handler(&StatusWriter{ResponseWriter: w}, r)
45
+
})
46
+
}
47
+
48
+
func (m *Mux) Use(middlewares ...Middleware) {
49
+
for _, middle := range middlewares {
50
+
m.middlewares = append(m.middlewares, func(next Handler) Handler {
51
+
return func(w *StatusWriter, r *http.Request) {
52
+
middle(next, w, r)
53
+
}
54
+
})
55
+
}
56
+
}
57
+
58
+
func MiddlewareLog(cancelCause func(context.Context) context.CancelCauseFunc, logNotFound, logBadRequest bool) Middleware {
59
+
return func(next Handler, w *StatusWriter, r *http.Request) {
60
+
log := slog.With("uri", r.RequestURI)
61
+
now := time.Now()
62
+
defer func() {
63
+
if err := recover(); err != nil {
64
+
w.WriteHeader(http.StatusInternalServerError)
65
+
log.Error("panic!", "error", err, "duration", time.Since(now))
66
+
switch e := err.(type) {
67
+
case error:
68
+
cancelCause(r.Context())(e)
69
+
case string:
70
+
cancelCause(r.Context())(errors.New(e))
71
+
default:
72
+
log.Warn(
73
+
"cannot set cancel cause, because error type is not supported",
74
+
"type", fmt.Sprintf("%T", e),
75
+
)
76
+
}
77
+
}
78
+
}()
79
+
80
+
next(w, r)
81
+
82
+
log = log.With("status", w.Code, "duration", time.Since(now))
83
+
if w.Code < 400 {
84
+
log.Debug("handled")
85
+
} else if w.Code < 500 {
86
+
level := slog.LevelDebug
87
+
if (w.Code == http.StatusNotFound && logNotFound) ||
88
+
(w.Code == http.StatusBadRequest && logBadRequest) ||
89
+
(w.Code != http.StatusNotFound && w.Code != http.StatusBadRequest) {
90
+
level = slog.LevelWarn
91
+
}
92
+
log.Log(context.Background(), level, "invalid request")
93
+
} else {
94
+
log.Error("error while handling request")
95
+
}
96
+
}
97
+
}
+5
-4
cmd/lasad/server/context.go
+5
-4
cmd/lasad/server/context.go
+6
-4
cmd/lasad/server/limiter.go
+6
-4
cmd/lasad/server/limiter.go
···
7
7
"strings"
8
8
"sync"
9
9
"time"
10
+
11
+
"tangled.org/anhgelus.world/lasa/cmd/internal"
10
12
)
11
13
12
14
type limited struct {
···
35
37
return ok && tm.time.Unix() > time.Now().Unix()
36
38
}
37
39
38
-
func (l *Limiter) handle(w *statusWriter, r *http.Request, log *slog.Logger) {
40
+
func (l *Limiter) handle(w *internal.StatusWriter, r *http.Request) {
39
41
ip, _, _ := strings.Cut(r.RemoteAddr, ":")
40
-
if w.code != http.StatusNotFound && w.code != http.StatusBadRequest && w.code != http.StatusTooManyRequests {
42
+
if w.Code != http.StatusNotFound && w.Code != http.StatusBadRequest && w.Code != http.StatusTooManyRequests {
41
43
return
42
44
}
43
45
var level uint = 0
44
-
if w.code == http.StatusBadRequest {
46
+
if w.Code == http.StatusBadRequest {
45
47
level = 1
46
48
}
47
49
···
58
60
if dur == 0 {
59
61
return
60
62
}
61
-
log.Info("rate limiting", "ip", ip, "for", dur)
63
+
slog.Info("rate limiting", "ip", ip, "for", dur, "uri", r.RequestURI)
62
64
tm.time = time.Now().Add(dur)
63
65
64
66
go func(l *Limiter, dur time.Duration, tm *limited, ip string) {
+22
-61
cmd/lasad/server/run.go
+22
-61
cmd/lasad/server/run.go
···
5
5
"embed"
6
6
"errors"
7
7
"fmt"
8
-
"log/slog"
9
8
"net/http"
10
9
"strings"
11
10
"time"
12
11
13
12
glide "github.com/valkey-io/valkey-glide/go/v2"
14
13
"tangled.org/anhgelus.world/lasa"
14
+
"tangled.org/anhgelus.world/lasa/cmd/internal"
15
15
"tangled.org/anhgelus.world/lasa/cmd/lasad/config"
16
16
"tangled.org/anhgelus.world/xrpc"
17
17
)
···
74
74
w.Write(b)
75
75
})
76
76
77
-
return http.ListenAndServe(fmt.Sprintf(":%d", cfg.Port), middlewares(mux, ctx))
78
-
}
79
-
80
-
type statusWriter struct {
81
-
http.ResponseWriter
82
-
code int
83
-
}
84
-
85
-
func (w *statusWriter) WriteHeader(statusCode int) {
86
-
w.ResponseWriter.WriteHeader(statusCode)
87
-
w.code = statusCode
88
-
}
89
-
90
-
func middlewares(h http.Handler, parent context.Context) http.Handler {
91
-
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
77
+
m := internal.NewMux(mux)
78
+
m.Use(func(next internal.Handler, w *internal.StatusWriter, r *http.Request) {
79
+
// not found favicon
92
80
if strings.HasPrefix(r.RequestURI, "/favicon.ico") {
93
81
w.WriteHeader(http.StatusNotFound)
94
82
return
95
83
}
96
-
84
+
next(w, r)
85
+
})
86
+
m.Use(func(next internal.Handler, w *internal.StatusWriter, r *http.Request) {
97
87
// timeouts request handling
98
-
ctx, cancel := context.WithTimeoutCause(parent, 15*time.Second, errors.New("handling timeouts"))
88
+
ctx, cancel := context.WithTimeoutCause(ctx, 15*time.Second, errors.New("handling timeouts"))
99
89
defer cancel()
100
90
101
91
var cancelCause context.CancelCauseFunc
102
92
ctx, cancelCause = context.WithCancelCause(ctx)
103
93
defer cancelCause(errors.New("handling finished"))
104
94
95
+
ctx = context.WithValue(ctx, keyCancelCause, cancelCause)
96
+
97
+
next(w, r.WithContext(ctx))
98
+
})
99
+
m.Use(internal.MiddlewareLog(func(ctx context.Context) context.CancelCauseFunc {
100
+
return ctx.Value(keyCancelCause).(context.CancelCauseFunc)
101
+
}, cfg.LogNotFound, cfg.LogBadRequest))
102
+
m.Use(func(next internal.Handler, w *internal.StatusWriter, r *http.Request) {
103
+
// rate limits
105
104
limiter := ctx.Value(keyLimiter).(*Limiter)
106
-
status := &statusWriter{w, http.StatusOK}
107
-
log := slog.With("uri", r.RequestURI)
108
105
if limiter.isLimited(r) {
109
-
status.WriteHeader(http.StatusTooManyRequests)
110
-
limiter.handle(status, r, log.With("status", http.StatusTooManyRequests))
106
+
w.WriteHeader(http.StatusTooManyRequests)
111
107
return
112
108
}
109
+
next(w, r)
110
+
limiter.handle(w, r)
111
+
})
113
112
114
-
now := time.Now()
115
-
defer func() {
116
-
if err := recover(); err != nil {
117
-
w.WriteHeader(http.StatusInternalServerError)
118
-
log.Error("panic!", "error", err, "duration", time.Since(now))
119
-
switch e := err.(type) {
120
-
case error:
121
-
cancelCause(e)
122
-
case string:
123
-
cancelCause(errors.New(e))
124
-
default:
125
-
log.Warn(
126
-
"cannot set cancel cause, because error type is not supported",
127
-
"type", fmt.Sprintf("%T", e),
128
-
)
129
-
}
130
-
}
131
-
}()
132
-
133
-
h.ServeHTTP(status, r.WithContext(ctx))
134
-
135
-
log = log.With("status", status.code, "duration", time.Since(now))
136
-
if status.code < 400 {
137
-
log.Debug("handled")
138
-
} else if status.code < 500 {
139
-
cfg := ctx.Value(keyCfg).(*config.Config)
140
-
level := slog.LevelDebug
141
-
if (status.code == http.StatusNotFound && cfg.LogNotFound) ||
142
-
(status.code == http.StatusBadRequest && cfg.LogBadRequest) ||
143
-
(status.code != http.StatusNotFound && status.code != http.StatusBadRequest) {
144
-
level = slog.LevelWarn
145
-
}
146
-
log.Log(context.Background(), level, "invalid request")
147
-
} else {
148
-
log.Error("error while handling request")
149
-
}
150
-
151
-
limiter.handle(status, r, log)
152
-
})
113
+
return http.ListenAndServe(fmt.Sprintf(":%d", cfg.Port), m.Handle())
153
114
}