addrkit: deterministic address resolution service (Go)
- domain: address normalization/parsing, matching with explicit statuses - adapters: SQLite registry store, ArcGIS outages client, optional LLM rewriter, HTTP API - seed pipeline with duplicate handling and unparsed reporting - CLI (seed/verify/resolve), systemd unit, OpenAPI contract, tests
This commit is contained in:
Maksim Totmin
@@ -0,0 +1,101 @@
|
||||
package httpapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"time"
|
||||
)
|
||||
|
||||
type ctxKey int
|
||||
|
||||
const requestIDKey ctxKey = 0
|
||||
|
||||
// requestIDMW assigns a request id (for correlation in logs) and echoes it.
|
||||
func requestIDMW(next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
id := r.Header.Get("X-Request-Id")
|
||||
if id == "" {
|
||||
id = newRequestID()
|
||||
}
|
||||
w.Header().Set("X-Request-Id", id)
|
||||
ctx := context.WithValue(r.Context(), requestIDKey, id)
|
||||
next.ServeHTTP(w, r.WithContext(ctx))
|
||||
})
|
||||
}
|
||||
|
||||
// loggingMW logs one line per request with method, path, status and duration.
|
||||
// It records no addresses (PII) — only routing metadata.
|
||||
func loggingMW(log *slog.Logger) func(http.Handler) http.Handler {
|
||||
return func(next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
start := time.Now()
|
||||
rec := &statusRecorder{ResponseWriter: w, status: http.StatusOK}
|
||||
next.ServeHTTP(rec, r)
|
||||
log.Info("http",
|
||||
"method", r.Method,
|
||||
"path", r.URL.Path,
|
||||
"status", rec.status,
|
||||
"duration_ms", time.Since(start).Milliseconds(),
|
||||
"request_id", requestIDFrom(r.Context()),
|
||||
)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// timeoutMW bounds the whole request, so a slow portal cannot tie up a
|
||||
// goroutine indefinitely.
|
||||
func timeoutMW(d time.Duration) func(http.Handler) http.Handler {
|
||||
return func(next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
ctx, cancel := context.WithTimeout(r.Context(), d)
|
||||
defer cancel()
|
||||
next.ServeHTTP(w, r.WithContext(ctx))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// recoverMW turns a panic into a 500 instead of crashing the process.
|
||||
func recoverMW(log *slog.Logger) func(http.Handler) http.Handler {
|
||||
return func(next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
defer func() {
|
||||
if rec := recover(); rec != nil {
|
||||
log.Error("panic", "value", rec, "request_id", requestIDFrom(r.Context()))
|
||||
writeError(w, http.StatusInternalServerError, "internal error")
|
||||
}
|
||||
}()
|
||||
next.ServeHTTP(w, r)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func requestIDFrom(ctx context.Context) string {
|
||||
if v, ok := ctx.Value(requestIDKey).(string); ok {
|
||||
return v
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
type statusRecorder struct {
|
||||
http.ResponseWriter
|
||||
status int
|
||||
}
|
||||
|
||||
func (s *statusRecorder) WriteHeader(code int) {
|
||||
s.status = code
|
||||
s.ResponseWriter.WriteHeader(code)
|
||||
}
|
||||
|
||||
// newRequestID returns a short random hex id. crypto/rand never fails in
|
||||
// practice; on the off chance it does we fall back to a constant so the
|
||||
// request still completes.
|
||||
func newRequestID() string {
|
||||
var b [16]byte
|
||||
if _, err := rand.Read(b[:]); err != nil {
|
||||
return "unknown"
|
||||
}
|
||||
return hex.EncodeToString(b[:])
|
||||
}
|
||||
@@ -0,0 +1,165 @@
|
||||
// Package httpapi exposes the addrkit domain over a small, versioned JSON API.
|
||||
//
|
||||
// It contains no business logic: handlers decode requests, call the resolver
|
||||
// and the outages source, and encode responses. Cross-cutting concerns
|
||||
// (timeouts, recovery, request ids, logging) live in middleware.
|
||||
package httpapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"git.totmin.ru/en2zmax/addrkit/internal/domain/address"
|
||||
"git.totmin.ru/en2zmax/addrkit/internal/domain/match"
|
||||
"git.totmin.ru/en2zmax/addrkit/internal/domain/outages"
|
||||
)
|
||||
|
||||
// Resolver is the read side of the domain (implemented by *match.Resolver).
|
||||
type Resolver interface {
|
||||
Resolve(ctx context.Context, raw string) match.Result
|
||||
Version() string
|
||||
Ready() bool
|
||||
}
|
||||
|
||||
// Server wires the domain to HTTP.
|
||||
type Server struct {
|
||||
resolver Resolver
|
||||
outages outages.Source
|
||||
log *slog.Logger
|
||||
now func() time.Time
|
||||
loc *time.Location
|
||||
timeout time.Duration
|
||||
}
|
||||
|
||||
// New builds a server. loc controls how outage dates are interpreted; it
|
||||
// should be the operational timezone (Asia/Novosibirsk).
|
||||
func New(resolver Resolver, src outages.Source, log *slog.Logger, loc *time.Location) *Server {
|
||||
if loc == nil {
|
||||
loc = time.UTC
|
||||
}
|
||||
return &Server{
|
||||
resolver: resolver,
|
||||
outages: src,
|
||||
log: log,
|
||||
now: time.Now,
|
||||
loc: loc,
|
||||
timeout: 10 * time.Second,
|
||||
}
|
||||
}
|
||||
|
||||
// Handler returns the fully wrapped HTTP handler.
|
||||
func (s *Server) Handler() http.Handler {
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("POST /v1/resolve", s.handleResolve)
|
||||
mux.HandleFunc("POST /v1/outages", s.handleOutages)
|
||||
mux.HandleFunc("GET /v1/healthz", s.handleHealthz)
|
||||
mux.HandleFunc("GET /v1/readyz", s.handleReadyz)
|
||||
mux.HandleFunc("GET /v1/version", s.handleVersion)
|
||||
|
||||
return recoverMW(s.log)(requestIDMW(loggingMW(s.log)(timeoutMW(s.timeout)(mux))))
|
||||
}
|
||||
|
||||
type resolveRequest struct {
|
||||
Address string `json:"address"`
|
||||
}
|
||||
|
||||
func (s *Server) handleResolve(w http.ResponseWriter, r *http.Request) {
|
||||
req, ok := s.decode(w, r)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, s.resolver.Resolve(r.Context(), req.Address))
|
||||
}
|
||||
|
||||
type outagesResponse struct {
|
||||
Status address.Status `json:"status"`
|
||||
Address string `json:"address,omitempty"`
|
||||
District string `json:"district,omitempty"`
|
||||
Outages []outages.Outage `json:"outages,omitempty"`
|
||||
AsOf string `json:"as_of,omitempty"`
|
||||
DataVersion string `json:"data_version,omitempty"`
|
||||
}
|
||||
|
||||
func (s *Server) handleOutages(w http.ResponseWriter, r *http.Request) {
|
||||
req, ok := s.decode(w, r)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
ctx := r.Context()
|
||||
res := s.resolver.Resolve(ctx, req.Address)
|
||||
|
||||
out := outagesResponse{
|
||||
Status: res.Status,
|
||||
Address: res.Address,
|
||||
District: res.District,
|
||||
DataVersion: res.DataVersion,
|
||||
}
|
||||
if res.Status != address.StatusFound {
|
||||
// No canonical address => nothing to query. Return the resolver's
|
||||
// verdict so the caller can ask for clarification.
|
||||
writeJSON(w, http.StatusOK, out)
|
||||
return
|
||||
}
|
||||
|
||||
raw, err := s.outages.Query(ctx, res.Address)
|
||||
if err != nil {
|
||||
if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
|
||||
out.Status = address.StatusPortalTimeout
|
||||
} else {
|
||||
out.Status = address.StatusInfra
|
||||
}
|
||||
writeJSON(w, http.StatusOK, out)
|
||||
return
|
||||
}
|
||||
|
||||
parsed := address.Parse(res.Address)
|
||||
current := outages.Current(raw, s.now(), s.loc)
|
||||
out.Outages = outages.MatchingHouse(current, parsed)
|
||||
out.AsOf = s.now().In(s.loc).Format(time.RFC3339)
|
||||
writeJSON(w, http.StatusOK, out)
|
||||
}
|
||||
|
||||
func (s *Server) handleHealthz(w http.ResponseWriter, _ *http.Request) {
|
||||
writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
|
||||
}
|
||||
|
||||
func (s *Server) handleReadyz(w http.ResponseWriter, _ *http.Request) {
|
||||
if !s.resolver.Ready() {
|
||||
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"status": "not_ready"})
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]string{"status": "ready"})
|
||||
}
|
||||
|
||||
func (s *Server) handleVersion(w http.ResponseWriter, _ *http.Request) {
|
||||
writeJSON(w, http.StatusOK, map[string]string{"data_version": s.resolver.Version()})
|
||||
}
|
||||
|
||||
// decode reads the JSON body into a resolveRequest, writing a 400 on failure.
|
||||
func (s *Server) decode(w http.ResponseWriter, r *http.Request) (resolveRequest, bool) {
|
||||
var req resolveRequest
|
||||
r.Body = http.MaxBytesReader(w, r.Body, 64*1024)
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
writeError(w, http.StatusBadRequest, "invalid json body")
|
||||
return resolveRequest{}, false
|
||||
}
|
||||
if req.Address == "" {
|
||||
writeError(w, http.StatusBadRequest, "address is required")
|
||||
return resolveRequest{}, false
|
||||
}
|
||||
return req, true
|
||||
}
|
||||
|
||||
func writeJSON(w http.ResponseWriter, status int, v any) {
|
||||
w.Header().Set("Content-Type", "application/json; charset=utf-8")
|
||||
w.WriteHeader(status)
|
||||
_ = json.NewEncoder(w).Encode(v)
|
||||
}
|
||||
|
||||
func writeError(w http.ResponseWriter, status int, msg string) {
|
||||
writeJSON(w, status, map[string]any{"error": 1, "message": msg})
|
||||
}
|
||||
@@ -0,0 +1,128 @@
|
||||
package httpapi
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.totmin.ru/en2zmax/addrkit/internal/domain/address"
|
||||
"git.totmin.ru/en2zmax/addrkit/internal/domain/match"
|
||||
"git.totmin.ru/en2zmax/addrkit/internal/domain/outages"
|
||||
)
|
||||
|
||||
func testResolver(t *testing.T) *match.Resolver {
|
||||
t.Helper()
|
||||
var entries []address.Entry
|
||||
for _, a := range []string{"Станционная улица 1/1", "Искитимская улица 5"} {
|
||||
p := address.Parse(a)
|
||||
entries = append(entries, address.Entry{
|
||||
ID: a, Address: a, District: "Тестовый",
|
||||
Ordinal: p.Ordinal, NameKey: p.NameKey, StreetType: p.StreetType, StreetKey: p.StreetKey,
|
||||
House: p.House, HouseFrac: p.HouseFrac, HouseLetter: p.HouseLetter,
|
||||
})
|
||||
}
|
||||
return match.NewResolver(match.NewIndex("test-version", entries, match.DefaultOptions()), nil)
|
||||
}
|
||||
|
||||
type fakeSource struct {
|
||||
list []outages.Outage
|
||||
err error
|
||||
}
|
||||
|
||||
func (f fakeSource) Query(context.Context, string) ([]outages.Outage, error) {
|
||||
return f.list, f.err
|
||||
}
|
||||
|
||||
func newServer(t *testing.T, src fakeSource) *httptest.Server {
|
||||
t.Helper()
|
||||
srv := New(testResolver(t), src, slog.New(slog.NewTextHandler(io.Discard, nil)), time.UTC)
|
||||
ts := httptest.NewServer(srv.Handler())
|
||||
t.Cleanup(ts.Close)
|
||||
return ts
|
||||
}
|
||||
|
||||
func postJSON(t *testing.T, url, body string) *http.Response {
|
||||
t.Helper()
|
||||
resp, err := http.Post(url, "application/json", bytes.NewBufferString(body))
|
||||
if err != nil {
|
||||
t.Fatalf("post: %v", err)
|
||||
}
|
||||
return resp
|
||||
}
|
||||
|
||||
func TestResolveEndpoint(t *testing.T) {
|
||||
ts := newServer(t, fakeSource{})
|
||||
resp := postJSON(t, ts.URL+"/v1/resolve", `{"address":"2-я Станционная 1/1"}`)
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Fatalf("status = %d", resp.StatusCode)
|
||||
}
|
||||
var res match.Result
|
||||
if err := json.NewDecoder(resp.Body).Decode(&res); err != nil {
|
||||
t.Fatalf("decode: %v", err)
|
||||
}
|
||||
if res.Status != address.StatusFound || res.Address != "Станционная улица 1/1" {
|
||||
t.Fatalf("unexpected result: %+v", res)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOutagesEndpointFiltersHouse(t *testing.T) {
|
||||
src := fakeSource{list: []outages.Outage{
|
||||
{Address: "Станционная улица 1/1", System: "Теплоснабжение"},
|
||||
{Address: "Станционная улица 1/2", System: "Теплоснабжение"},
|
||||
}}
|
||||
ts := newServer(t, src)
|
||||
resp := postJSON(t, ts.URL+"/v1/outages", `{"address":"станционная 1/1"}`)
|
||||
defer resp.Body.Close()
|
||||
|
||||
var out outagesResponse
|
||||
if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
|
||||
t.Fatalf("decode: %v", err)
|
||||
}
|
||||
if out.Status != address.StatusFound {
|
||||
t.Fatalf("status = %q, want found", out.Status)
|
||||
}
|
||||
if len(out.Outages) != 1 || out.Outages[0].Address != "Станционная улица 1/1" {
|
||||
t.Fatalf("house filter failed: %+v", out.Outages)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveValidation(t *testing.T) {
|
||||
ts := newServer(t, fakeSource{})
|
||||
resp := postJSON(t, ts.URL+"/v1/resolve", `{}`)
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusBadRequest {
|
||||
t.Fatalf("status = %d, want 400", resp.StatusCode)
|
||||
}
|
||||
}
|
||||
|
||||
func TestReadyzAndVersion(t *testing.T) {
|
||||
ts := newServer(t, fakeSource{})
|
||||
|
||||
resp, err := http.Get(ts.URL + "/v1/readyz")
|
||||
if err != nil {
|
||||
t.Fatalf("readyz: %v", err)
|
||||
}
|
||||
resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Fatalf("readyz status = %d", resp.StatusCode)
|
||||
}
|
||||
|
||||
resp, err = http.Get(ts.URL + "/v1/version")
|
||||
if err != nil {
|
||||
t.Fatalf("version: %v", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
var v map[string]string
|
||||
_ = json.NewDecoder(resp.Body).Decode(&v)
|
||||
if v["data_version"] != "test-version" {
|
||||
t.Fatalf("version = %q", v["data_version"])
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user