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:
addrkit
@@ -0,0 +1,226 @@
|
||||
// Package arcgis is the adapter for the external outages portal exposed as an
|
||||
// ArcGIS MapServer /query endpoint.
|
||||
//
|
||||
// The portal only supports substring (LIKE) filtering, so precision is
|
||||
// guaranteed locally: whatever the portal returns is re-filtered by the
|
||||
// outages domain using the parsed house unit. The client adds the reliability
|
||||
// the portal lacks: a hard timeout, bounded retries with backoff and context
|
||||
// cancellation.
|
||||
package arcgis
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.totmin.ru/en2zmax/addrkit/internal/domain/outages"
|
||||
)
|
||||
|
||||
// Client is a bounded HTTP client for the portal.
|
||||
type Client struct {
|
||||
baseURL string
|
||||
http *http.Client
|
||||
retries int
|
||||
}
|
||||
|
||||
// New builds a client. timeout applies to a single attempt; retries is the
|
||||
// number of *additional* attempts after the first.
|
||||
func New(baseURL string, timeout time.Duration, retries int) *Client {
|
||||
return &Client{
|
||||
baseURL: baseURL,
|
||||
http: &http.Client{Timeout: timeout},
|
||||
retries: retries,
|
||||
}
|
||||
}
|
||||
|
||||
// Query asks the portal for outages matching the canonical address. The
|
||||
// returned records are raw portal rows; callers must still filter by house.
|
||||
func (c *Client) Query(ctx context.Context, canonical string) ([]outages.Outage, error) {
|
||||
params := url.Values{}
|
||||
params.Set("where", "address like '"+likePattern(canonical)+"'")
|
||||
params.Set("outFields", "")
|
||||
params.Set("returnGeometry", "false")
|
||||
params.Set("f", "json")
|
||||
|
||||
var lastErr error
|
||||
for attempt := 0; attempt <= c.retries; attempt++ {
|
||||
if attempt > 0 {
|
||||
backoff := time.Duration(attempt) * 150 * time.Millisecond
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
case <-time.After(backoff):
|
||||
}
|
||||
}
|
||||
rows, err := c.do(ctx, params)
|
||||
if err == nil {
|
||||
return rows, nil
|
||||
}
|
||||
lastErr = err
|
||||
}
|
||||
return nil, fmt.Errorf("portal query failed after %d attempts: %w", c.retries+1, lastErr)
|
||||
}
|
||||
|
||||
func (c *Client) do(ctx context.Context, params url.Values) ([]outages.Outage, error) {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.baseURL+"?"+params.Encode(), nil)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("build request: %w", err)
|
||||
}
|
||||
resp, err := c.http.Do(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return nil, fmt.Errorf("portal status %d", resp.StatusCode)
|
||||
}
|
||||
// Bound the response size to protect the service from a misbehaving portal.
|
||||
body, err := io.ReadAll(io.LimitReader(resp.Body, 4<<20))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return parseResponse(body)
|
||||
}
|
||||
|
||||
// portalResponse mirrors the subset of the ArcGIS /query envelope we use.
|
||||
type portalResponse struct {
|
||||
Features []struct {
|
||||
Value string `json:"value"`
|
||||
Attributes map[string]any `json:"attributes"`
|
||||
} `json:"features"`
|
||||
Fields []struct {
|
||||
Name string `json:"name"`
|
||||
Domain struct {
|
||||
CodedValues []struct {
|
||||
Code any `json:"code"`
|
||||
Name string `json:"name"`
|
||||
} `json:"codedValues"`
|
||||
} `json:"domain"`
|
||||
} `json:"fields"`
|
||||
}
|
||||
|
||||
func parseResponse(body []byte) ([]outages.Outage, error) {
|
||||
var pr portalResponse
|
||||
if err := json.Unmarshal(body, &pr); err != nil {
|
||||
return nil, fmt.Errorf("decode portal response: %w", err)
|
||||
}
|
||||
|
||||
// Build code -> name maps from the coded-value domains.
|
||||
maps := map[string]map[string]string{}
|
||||
for _, f := range pr.Fields {
|
||||
if len(f.Domain.CodedValues) == 0 {
|
||||
continue
|
||||
}
|
||||
m := make(map[string]string, len(f.Domain.CodedValues))
|
||||
for _, cv := range f.Domain.CodedValues {
|
||||
m[codeKey(cv.Code)] = cv.Name
|
||||
}
|
||||
maps[f.Name] = m
|
||||
}
|
||||
|
||||
out := make([]outages.Outage, 0, len(pr.Features))
|
||||
for _, feat := range pr.Features {
|
||||
o := outages.Outage{
|
||||
Address: addressOf(feat.Attributes, feat.Value),
|
||||
Type: lookup(maps, "type_id", feat.Attributes),
|
||||
System: lookup(maps, "system_id", feat.Attributes),
|
||||
Time: lookup(maps, "time_category", feat.Attributes),
|
||||
StartDate: stringOf(feat.Attributes["start_date"]),
|
||||
EndDate: stringOf(feat.Attributes["end_date"]),
|
||||
}
|
||||
o.Action = actionFor(o.System)
|
||||
out = append(out, o)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func addressOf(attrs map[string]any, fallback string) string {
|
||||
if v, ok := attrs["address"]; ok {
|
||||
if s := stringOf(v); s != "" {
|
||||
return s
|
||||
}
|
||||
}
|
||||
return fallback
|
||||
}
|
||||
|
||||
func lookup(maps map[string]map[string]string, field string, attrs map[string]any) string {
|
||||
m, ok := maps[field]
|
||||
if !ok {
|
||||
return stringOf(attrs[field])
|
||||
}
|
||||
return m[codeKey(attrs[field])]
|
||||
}
|
||||
|
||||
// codeKey normalises a coded value coming from JSON (float64) or a string.
|
||||
func codeKey(v any) string {
|
||||
switch t := v.(type) {
|
||||
case nil:
|
||||
return ""
|
||||
case string:
|
||||
return t
|
||||
case float64:
|
||||
return strconv.FormatInt(int64(t), 10)
|
||||
case json.Number:
|
||||
return t.String()
|
||||
default:
|
||||
return fmt.Sprintf("%v", v)
|
||||
}
|
||||
}
|
||||
|
||||
func stringOf(v any) string {
|
||||
if s, ok := v.(string); ok {
|
||||
return s
|
||||
}
|
||||
if v == nil {
|
||||
return ""
|
||||
}
|
||||
return fmt.Sprintf("%v", v)
|
||||
}
|
||||
|
||||
// actionFor maps a portal system name to the canonical action code used by the
|
||||
// scenario (kept for parity with the existing integration).
|
||||
func actionFor(system string) string {
|
||||
switch system {
|
||||
case "Горячее водоснабжение":
|
||||
return "waterWarm"
|
||||
case "Холодное водоснабжение и канализация", "Холодное водоснабжение":
|
||||
return "waterCold"
|
||||
case "Электроснабжение":
|
||||
return "electricitySupply"
|
||||
case "Газоснабжение":
|
||||
return "gasSupply"
|
||||
case "Теплоснабжение":
|
||||
return "heating"
|
||||
default:
|
||||
return ""
|
||||
}
|
||||
}
|
||||
|
||||
// likePattern builds the substring pattern the portal supports. Numeric tokens
|
||||
// are padded with spaces, matching the historical integration, so that "1" does
|
||||
// not incidentally match "10".
|
||||
func likePattern(canonical string) string {
|
||||
parts := strings.Fields(canonical)
|
||||
for i, part := range parts {
|
||||
if containsDigit(part) {
|
||||
parts[i] = " " + part + " "
|
||||
}
|
||||
}
|
||||
return "%" + strings.Join(parts, "%") + "%"
|
||||
}
|
||||
|
||||
func containsDigit(s string) bool {
|
||||
for _, r := range s {
|
||||
if r >= '0' && r <= '9' {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -0,0 +1,68 @@
|
||||
package arcgis
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
const envelope = `{
|
||||
"features": [
|
||||
{"value": "Станционная улица 1/1", "attributes": {
|
||||
"address": "Станционная улица 1/1",
|
||||
"type_id": 1, "system_id": 5, "time_category": 2,
|
||||
"start_date": "2026-10-01T00:00:00", "end_date": "2026-10-31T00:00:00"
|
||||
}}
|
||||
],
|
||||
"fields": [
|
||||
{"name": "type_id", "domain": {"codedValues": [{"code": 1, "name": "Плановое"}]}},
|
||||
{"name": "system_id", "domain": {"codedValues": [{"code": 5, "name": "Теплоснабжение"}]}},
|
||||
{"name": "time_category", "domain": {"codedValues": [{"code": 2, "name": "Текущие"}]}}
|
||||
]
|
||||
}`
|
||||
|
||||
func TestQueryParsesEnvelope(t *testing.T) {
|
||||
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if got := r.URL.Query().Get("f"); got != "json" {
|
||||
t.Errorf("missing f=json, got %q", got)
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_, _ = w.Write([]byte(envelope))
|
||||
}))
|
||||
defer ts.Close()
|
||||
|
||||
c := New(ts.URL, time.Second, 0)
|
||||
rows, err := c.Query(context.Background(), "Станционная улица 1/1")
|
||||
if err != nil {
|
||||
t.Fatalf("Query: %v", err)
|
||||
}
|
||||
if len(rows) != 1 {
|
||||
t.Fatalf("rows = %d, want 1", len(rows))
|
||||
}
|
||||
got := rows[0]
|
||||
if got.Type != "Плановое" || got.System != "Теплоснабжение" || got.Time != "Текущие" {
|
||||
t.Errorf("coded values not resolved: %+v", got)
|
||||
}
|
||||
if got.Action != "heating" {
|
||||
t.Errorf("Action = %q, want heating", got.Action)
|
||||
}
|
||||
}
|
||||
|
||||
func TestQueryRetriesThenFails(t *testing.T) {
|
||||
var calls int
|
||||
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
calls++
|
||||
http.Error(w, "boom", http.StatusInternalServerError)
|
||||
}))
|
||||
defer ts.Close()
|
||||
|
||||
c := New(ts.URL, time.Second, 2)
|
||||
if _, err := c.Query(context.Background(), "Станционная улица 1/1"); err == nil {
|
||||
t.Fatal("expected error")
|
||||
}
|
||||
if calls != 3 {
|
||||
t.Errorf("calls = %d, want 3 (1 + 2 retries)", calls)
|
||||
}
|
||||
}
|
||||
@@ -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"])
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,137 @@
|
||||
// Package llm implements the optional, disabled-by-default query rewriter.
|
||||
//
|
||||
// It talks to any OpenAI-compatible chat completions endpoint (for example the
|
||||
// on-prem vLLM already used by AMS). Its contract is deliberately narrow: it
|
||||
// may only propose alternative spellings of the user's address. It never
|
||||
// returns the answer, and its output is validated by the deterministic
|
||||
// resolver, so a misbehaving model can at worst waste a request — it cannot
|
||||
// make the service answer with a wrong address.
|
||||
package llm
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.totmin.ru/en2zmax/addrkit/internal/config"
|
||||
)
|
||||
|
||||
// Rewriter asks a model for normalised variants of an address.
|
||||
type Rewriter struct {
|
||||
endpoint string
|
||||
token string
|
||||
model string
|
||||
http *http.Client
|
||||
}
|
||||
|
||||
// New builds a rewriter from configuration. The URL is treated as a base; the
|
||||
// OpenAI path is appended if it is not already present.
|
||||
func New(cfg config.LLMConfig) *Rewriter {
|
||||
base := strings.TrimRight(cfg.URL, "/")
|
||||
if !strings.HasSuffix(base, "/chat/completions") {
|
||||
base += "/v1/chat/completions"
|
||||
}
|
||||
timeout := time.Duration(cfg.TimeoutSeconds) * time.Second
|
||||
if timeout <= 0 {
|
||||
timeout = 3 * time.Second
|
||||
}
|
||||
return &Rewriter{
|
||||
endpoint: base,
|
||||
token: cfg.Token,
|
||||
model: cfg.Model,
|
||||
http: &http.Client{Timeout: timeout},
|
||||
}
|
||||
}
|
||||
|
||||
const systemPrompt = `Ты — нормализатор адресов. Тебе дают адрес, написанный как угодно.
|
||||
Верни ТОЛЬКО JSON вида {"variants":["...","..."]} — до 3 вариантов записи того же адреса
|
||||
в формате "<название улицы> <дом>[/<дробь>][ <литера>][ к<корпус>][ стр.<строение>]".
|
||||
Не выдумывай улицы и дома, которых нет во входе. Если не уверен — верни пустой список.`
|
||||
|
||||
type chatRequest struct {
|
||||
Model string `json:"model"`
|
||||
Messages []chatMessage `json:"messages"`
|
||||
Temperature float64 `json:"temperature"`
|
||||
MaxTokens int `json:"max_tokens"`
|
||||
}
|
||||
|
||||
type chatMessage struct {
|
||||
Role string `json:"role"`
|
||||
Content string `json:"content"`
|
||||
}
|
||||
|
||||
type chatResponse struct {
|
||||
Choices []struct {
|
||||
Message chatMessage `json:"message"`
|
||||
} `json:"choices"`
|
||||
}
|
||||
|
||||
// Rewrite returns up to three normalised variants of raw.
|
||||
func (r *Rewriter) Rewrite(ctx context.Context, raw string) ([]string, error) {
|
||||
payload, err := json.Marshal(chatRequest{
|
||||
Model: r.model,
|
||||
Messages: []chatMessage{
|
||||
{Role: "system", Content: systemPrompt},
|
||||
{Role: "user", Content: raw},
|
||||
},
|
||||
Temperature: 0,
|
||||
MaxTokens: 256,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, r.endpoint, bytes.NewReader(payload))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
if r.token != "" && r.token != "EMPTY" {
|
||||
req.Header.Set("Authorization", "Bearer "+r.token)
|
||||
}
|
||||
|
||||
resp, err := r.http.Do(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return nil, fmt.Errorf("llm status %d", resp.StatusCode)
|
||||
}
|
||||
|
||||
var cr chatResponse
|
||||
if err := json.NewDecoder(resp.Body).Decode(&cr); err != nil {
|
||||
return nil, fmt.Errorf("decode llm response: %w", err)
|
||||
}
|
||||
if len(cr.Choices) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
return parseVariants(cr.Choices[0].Message.Content), nil
|
||||
}
|
||||
|
||||
// parseVariants extracts the {"variants":[...]} object from a model reply that
|
||||
// may be wrapped in prose or code fences.
|
||||
func parseVariants(content string) []string {
|
||||
start := strings.Index(content, "{")
|
||||
end := strings.LastIndex(content, "}")
|
||||
if start < 0 || end <= start {
|
||||
return nil
|
||||
}
|
||||
var parsed struct {
|
||||
Variants []string `json:"variants"`
|
||||
}
|
||||
if err := json.Unmarshal([]byte(content[start:end+1]), &parsed); err != nil {
|
||||
return nil
|
||||
}
|
||||
out := make([]string, 0, len(parsed.Variants))
|
||||
for _, v := range parsed.Variants {
|
||||
if s := strings.TrimSpace(v); s != "" {
|
||||
out = append(out, s)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,169 @@
|
||||
// Package registrystore persists the canonical registry as a single,
|
||||
// self-contained SQLite file. It is intentionally boring: one file, no server,
|
||||
// built once by the seed pipeline and opened read-only by the service.
|
||||
package registrystore
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"os"
|
||||
|
||||
_ "modernc.org/sqlite" // registers the "sqlite" driver
|
||||
|
||||
"git.totmin.ru/en2zmax/addrkit/internal/domain/address"
|
||||
)
|
||||
|
||||
const schema = `
|
||||
CREATE TABLE IF NOT EXISTS meta (
|
||||
key TEXT PRIMARY KEY,
|
||||
value TEXT NOT NULL
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS address (
|
||||
id TEXT PRIMARY KEY,
|
||||
address TEXT NOT NULL,
|
||||
district TEXT NOT NULL DEFAULT '',
|
||||
ordinal TEXT NOT NULL DEFAULT '',
|
||||
name_key TEXT NOT NULL DEFAULT '',
|
||||
name_display TEXT NOT NULL DEFAULT '',
|
||||
street_type TEXT NOT NULL DEFAULT '',
|
||||
street_key TEXT NOT NULL,
|
||||
house TEXT NOT NULL DEFAULT '',
|
||||
house_frac TEXT NOT NULL DEFAULT '',
|
||||
house_letter TEXT NOT NULL DEFAULT '',
|
||||
building TEXT NOT NULL DEFAULT '',
|
||||
building_frac TEXT NOT NULL DEFAULT '',
|
||||
building_letter TEXT NOT NULL DEFAULT '',
|
||||
structure TEXT NOT NULL DEFAULT ''
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_address_street ON address(street_key);
|
||||
`
|
||||
|
||||
// Save writes entries to path atomically: it builds a temporary file and
|
||||
// renames it into place, so readers never observe a half-written index.
|
||||
//
|
||||
// The build uses the rollback journal (not WAL) on purpose, so that the result
|
||||
// is a single file with no companion -wal/-shm that a rename would orphan.
|
||||
func Save(path, version string, entries []address.Entry) error {
|
||||
tmp := path + ".tmp"
|
||||
removeDBFiles(tmp)
|
||||
|
||||
db, err := sql.Open("sqlite", "file:"+tmp+
|
||||
"?_pragma=journal_mode(DELETE)"+
|
||||
"&_pragma=synchronous(FULL)"+
|
||||
"&_pragma=busy_timeout(5000)")
|
||||
if err != nil {
|
||||
return fmt.Errorf("open temp db: %w", err)
|
||||
}
|
||||
if _, err := db.Exec(schema); err != nil {
|
||||
db.Close()
|
||||
return fmt.Errorf("create schema: %w", err)
|
||||
}
|
||||
if err := insertAll(db, version, entries); err != nil {
|
||||
db.Close()
|
||||
return err
|
||||
}
|
||||
if err := db.Close(); err != nil {
|
||||
return fmt.Errorf("close temp db: %w", err)
|
||||
}
|
||||
if err := os.Rename(tmp, path); err != nil {
|
||||
return fmt.Errorf("publish db: %w", err)
|
||||
}
|
||||
// Drop only the temp file's journal siblings; the published base file must
|
||||
// survive.
|
||||
removeSidecars(tmp)
|
||||
return nil
|
||||
}
|
||||
|
||||
func insertAll(db *sql.DB, version string, entries []address.Entry) error {
|
||||
tx, err := db.Begin()
|
||||
if err != nil {
|
||||
return fmt.Errorf("begin: %w", err)
|
||||
}
|
||||
defer tx.Rollback()
|
||||
|
||||
if _, err := tx.Exec(`INSERT INTO meta(key, value) VALUES ('version', ?)`, version); err != nil {
|
||||
return fmt.Errorf("insert meta: %w", err)
|
||||
}
|
||||
|
||||
stmt, err := tx.Prepare(`INSERT INTO address (
|
||||
id, address, district, ordinal, name_key, name_display, street_type, street_key,
|
||||
house, house_frac, house_letter, building, building_frac, building_letter, structure
|
||||
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`)
|
||||
if err != nil {
|
||||
return fmt.Errorf("prepare insert: %w", err)
|
||||
}
|
||||
defer stmt.Close()
|
||||
|
||||
for i, e := range entries {
|
||||
id := e.ID
|
||||
if id == "" {
|
||||
// Guarantee a primary key even for sources without ids.
|
||||
id = fmt.Sprintf("row-%d", i)
|
||||
}
|
||||
if _, err := stmt.Exec(
|
||||
id, e.Address, e.District, e.Ordinal, e.NameKey, e.NameDisplay, e.StreetType, e.StreetKey,
|
||||
e.House, e.HouseFrac, e.HouseLetter, e.Building, e.BuildingFrac, e.BuildingLetter, e.Structure,
|
||||
); err != nil {
|
||||
return fmt.Errorf("insert %q: %w", e.Address, err)
|
||||
}
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return fmt.Errorf("commit: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Load opens path read-only and returns the data version and all entries.
|
||||
func Load(path string) (version string, entries []address.Entry, err error) {
|
||||
if _, err := os.Stat(path); err != nil {
|
||||
return "", nil, fmt.Errorf("stat db: %w", err)
|
||||
}
|
||||
db, err := sql.Open("sqlite", "file:"+path+
|
||||
"?mode=ro"+
|
||||
"&_pragma=query_only(1)"+
|
||||
"&_pragma=busy_timeout(5000)")
|
||||
if err != nil {
|
||||
return "", nil, fmt.Errorf("open db: %w", err)
|
||||
}
|
||||
defer db.Close()
|
||||
|
||||
if err := db.QueryRow(`SELECT value FROM meta WHERE key='version'`).Scan(&version); err != nil {
|
||||
return "", nil, fmt.Errorf("read version: %w", err)
|
||||
}
|
||||
|
||||
rows, err := db.Query(`SELECT id, address, district, ordinal, name_key, name_display,
|
||||
street_type, street_key, house, house_frac, house_letter, building, building_frac,
|
||||
building_letter, structure FROM address`)
|
||||
if err != nil {
|
||||
return "", nil, fmt.Errorf("query: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
for rows.Next() {
|
||||
var e address.Entry
|
||||
if err := rows.Scan(
|
||||
&e.ID, &e.Address, &e.District, &e.Ordinal, &e.NameKey, &e.NameDisplay,
|
||||
&e.StreetType, &e.StreetKey, &e.House, &e.HouseFrac, &e.HouseLetter,
|
||||
&e.Building, &e.BuildingFrac, &e.BuildingLetter, &e.Structure,
|
||||
); err != nil {
|
||||
return "", nil, fmt.Errorf("scan: %w", err)
|
||||
}
|
||||
entries = append(entries, e)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return "", nil, fmt.Errorf("rows: %w", err)
|
||||
}
|
||||
return version, entries, nil
|
||||
}
|
||||
|
||||
// removeDBFiles deletes a database file and its journal siblings.
|
||||
func removeDBFiles(base string) {
|
||||
_ = os.Remove(base)
|
||||
removeSidecars(base)
|
||||
}
|
||||
|
||||
// removeSidecars deletes only the -wal/-shm companions of a database file.
|
||||
func removeSidecars(base string) {
|
||||
_ = os.Remove(base + "-wal")
|
||||
_ = os.Remove(base + "-shm")
|
||||
}
|
||||
@@ -0,0 +1,59 @@
|
||||
package registrystore
|
||||
|
||||
import (
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"git.totmin.ru/en2zmax/addrkit/internal/domain/address"
|
||||
)
|
||||
|
||||
func TestSaveLoadRoundTrip(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), "registry.db")
|
||||
|
||||
addr := "Станционная улица 1/1"
|
||||
p := address.Parse(addr)
|
||||
entries := []address.Entry{{
|
||||
ID: "a1", Address: addr, District: "Ленинский",
|
||||
Ordinal: p.Ordinal, NameKey: p.NameKey, StreetType: p.StreetType, StreetKey: p.StreetKey,
|
||||
House: p.House, HouseFrac: p.HouseFrac,
|
||||
}}
|
||||
|
||||
if err := Save(path, "2026-10-09", entries); err != nil {
|
||||
t.Fatalf("Save: %v", err)
|
||||
}
|
||||
|
||||
version, got, err := Load(path)
|
||||
if err != nil {
|
||||
t.Fatalf("Load: %v", err)
|
||||
}
|
||||
if version != "2026-10-09" {
|
||||
t.Errorf("version = %q", version)
|
||||
}
|
||||
if len(got) != 1 {
|
||||
t.Fatalf("entries = %d, want 1", len(got))
|
||||
}
|
||||
if got[0].Address != addr || got[0].District != "Ленинский" || got[0].StreetKey != p.StreetKey {
|
||||
t.Errorf("round-tripped entry mismatch: %+v", got[0])
|
||||
}
|
||||
}
|
||||
|
||||
// TestSaveIsAtomicOverwrite verifies that re-seeding replaces the previous data
|
||||
// rather than merging with it.
|
||||
func TestSaveIsAtomicOverwrite(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), "registry.db")
|
||||
|
||||
if err := Save(path, "v1", []address.Entry{{ID: "x", Address: "Первая улица 1"}}); err != nil {
|
||||
t.Fatalf("Save v1: %v", err)
|
||||
}
|
||||
if err := Save(path, "v2", []address.Entry{{ID: "y", Address: "Вторая улица 2"}}); err != nil {
|
||||
t.Fatalf("Save v2: %v", err)
|
||||
}
|
||||
|
||||
version, got, err := Load(path)
|
||||
if err != nil {
|
||||
t.Fatalf("Load: %v", err)
|
||||
}
|
||||
if version != "v2" || len(got) != 1 || got[0].Address != "Вторая улица 2" {
|
||||
t.Fatalf("unexpected state after overwrite: version=%q entries=%+v", version, got)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user