28705447299

Committed 04 Jul 2026 11:57AM UTC coverage: 97.686% (-0.05%) from 97.737%

Build # 28705447299

Build Type

push

github

Committed by

web-flow

Commit Message

Index rebuild #2 (#159)

* Index rebuild #2

* test

Coverage Stats

2732 of 2759 branches covered (99.02%)

Branch coverage included in aggregate %.

192 of 207 new or added lines in 6 files covered. (92.75%)

18670 of 19150 relevant lines covered (97.49%)

70.88 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.26

/internal/handler/admin.go

package handler

import (
        "context"
        "net/http"
        "time"

        "github.com/DigitalTolk/ex/internal/middleware"
        "github.com/DigitalTolk/ex/internal/model"
        "github.com/DigitalTolk/ex/internal/search"
        "github.com/DigitalTolk/ex/internal/service"
)

// SearchStatusReporter is the slim view AdminHandler needs to render
// the search status panel: cluster health + per-index docs/size. The
// concrete *search.Client satisfies this directly.
type SearchStatusReporter interface {
        ClusterHealth(ctx context.Context) (map[string]any, error)
        IndexStats(ctx context.Context) ([]search.IndexStat, error)
}

// MappingRebuildController is the slim view AdminHandler needs to trigger and
// report the cluster-coordinated users/channels mapping rebuild. The concrete
// *search.MappingRebuilder satisfies it; tests inject a fake.
type MappingRebuildController interface {
        Start(ctx context.Context, now func() int64) (bool, error)
        Status(ctx context.Context) (search.MappingRebuildStatus, error)
        SchemaVersions(ctx context.Context) ([]search.SchemaVersionInfo, error)
}

// AdminHandler exposes admin-only endpoints for workspace configuration.
// Authorization is enforced inside each handler — `middleware.Auth` only
// confirms the caller is signed in, not that they're an admin.
type AdminHandler struct {
        settings   *service.SettingsService
        searchSt   SearchStatusReporter
        reindexer  *search.Reindexer
        mappingReb MappingRebuildController
}

// NewAdminHandler constructs an AdminHandler.
func NewAdminHandler(settings *service.SettingsService) *AdminHandler {
        return &AdminHandler{settings: settings}
}

// SetSearch wires the optional search status reporter + reindexer.
// Production passes the live ones; tests can pass nil and the
// search-admin routes return 503.
func (h *AdminHandler) SetSearch(reporter SearchStatusReporter, reindexer *search.Reindexer) {
        h.searchSt = reporter
        h.reindexer = reindexer
}

// SetMappingRebuilder wires the optional cluster-coordinated users/channels
// mapping rebuild (RecreateUsersChannels behind a Redis lock + shared status).
// nil → the mapping-rebuild route answers 503, same as an unconfigured search.
func (h *AdminHandler) SetMappingRebuilder(reb MappingRebuildController) {
        h.mappingReb = reb
}

// settingsResponse is the wire shape returned from GetSettings. It
// extends model.WorkspaceSettings with a derived `giphyEnabled` flag.
// GIPHY requires API and media requests to be made directly by the
// client, so authenticated members receive the configured browser key
// when the picker is enabled.
type settingsResponse struct {
        MaxUploadBytes    int64    `json:"maxUploadBytes"`
        AllowedExtensions []string `json:"allowedExtensions"`
        GiphyAPIKey       string   `json:"giphyAPIKey,omitempty"`
        GiphyEnabled      bool     `json:"giphyEnabled"`
}

// GetSettings returns the effective workspace settings (with defaults
// applied for any field the admin hasn't overridden). Available to all
// authenticated users so the upload UI can show the limits before
// attempting a request — the write side is admin-only.
func (h *AdminHandler) GetSettings(w http.ResponseWriter, r *http.Request) {
        claims := middleware.ClaimsFromContext(r.Context())
        if claims == nil {
                writeError(w, http.StatusUnauthorized, "unauthorized", "authentication required")
                return
        }
        ws := h.settings.Effective(r.Context())
        resp := settingsResponse{
                MaxUploadBytes:    ws.MaxUploadBytes,
                AllowedExtensions: ws.AllowedExtensions,
                GiphyEnabled:      ws.GiphyAPIKey != "",
        }
        resp.GiphyAPIKey = ws.GiphyAPIKey
        writeJSON(w, http.StatusOK, resp)
}

// SearchStatus returns the search cluster's health, per-index doc
// counts/sizes, and the most recent reindex progress. Admin-only.
// Returns 503 with a structured payload (`configured: false`) when
// search isn't wired so the UI can render the panel without erroring.
func (h *AdminHandler) SearchStatus(w http.ResponseWriter, r *http.Request) {
        if !requireAdmin(w, r) {
                return
        }
        if h.searchSt == nil || h.reindexer == nil {
                writeJSON(w, http.StatusOK, JSON{"configured": false})
                return
        }
        ctx := r.Context()
        resp := JSON{"configured": true}
        if health, err := h.searchSt.ClusterHealth(ctx); err == nil && health != nil {
                resp["cluster"] = health
        } else if err != nil {
                resp["clusterError"] = err.Error()
        }
        if stats, err := h.searchSt.IndexStats(ctx); err == nil {
                resp["indices"] = stats
        } else {
                resp["indicesError"] = err.Error()
        }
        if rst, err := h.reindexer.Status(ctx); err == nil {
                resp["reindex"] = rst
        } else {
                resp["reindexError"] = err.Error()
        }
        if h.mappingReb != nil {
                if st, err := h.mappingReb.Status(ctx); err == nil {
                        resp["mappingRebuild"] = st
                } else {
                        resp["mappingRebuildError"] = err.Error()
                }
                if versions, err := h.mappingReb.SchemaVersions(ctx); err == nil {
                        resp["schemaVersions"] = versions
                } else {
                        resp["schemaVersionsError"] = err.Error()
                }
        }
        writeJSON(w, http.StatusOK, resp)
}

// StartSearchReindex kicks off a full rebuild in the background.
// Admin-only. Returns 202 if a fresh run started, 409 if one is
// already running, 503 if search isn't configured.
func (h *AdminHandler) StartSearchReindex(w http.ResponseWriter, r *http.Request) {
        if !requireAdmin(w, r) {
                return
        }
        if h.reindexer == nil {
                writeError(w, http.StatusServiceUnavailable, "search_disabled", "search is not configured")
                return
        }
        // Detach the request context so the goroutine survives the HTTP
        // response. Reindexes routinely outlive the request timeout —
        // admins poll status via SearchStatus afterwards.
        if !h.reindexer.Start(context.Background()) {
                writeError(w, http.StatusConflict, "already_running", "a reindex is already running")
                return
        }
        st, err := h.reindexer.Status(r.Context())
        if err != nil {
                // The run started; a status read-back failure is non-fatal — report
                // accepted with an empty snapshot, the panel will poll.
                writeJSON(w, http.StatusAccepted, search.ReindexProgress{Running: true})
                return
        }
        writeJSON(w, http.StatusAccepted, st)
}

// StartSearchMappingRebuild kicks off the cluster-coordinated users/channels
// mapping rebuild (staging index → atomic alias-swap) — the path that actually
// rolls a new analyzer onto an existing cluster. Admin-only. A Redis lock makes
// it single-flight across every instance: 202 if THIS call won the lock and
// started a run, 409 if another instance (or a still-cooling crashed run) holds
// it, 503 if search isn't configured, 500 on a coordination (Redis) error.
func (h *AdminHandler) StartSearchMappingRebuild(w http.ResponseWriter, r *http.Request) {
        if !requireAdmin(w, r) {
                return
        }
        if h.mappingReb == nil {
                writeError(w, http.StatusServiceUnavailable, "search_disabled", "search is not configured")
                return
        }
        // Detach from the request context so the lock/state writes and the detached
        // rebuild goroutine survive the HTTP response; admins poll status afterwards.
        started, err := h.mappingReb.Start(context.Background(), nowUnix)
        if err != nil {
                // Generic 500 — don't leak the Redis/coordination error to the client.
                writeError(w, http.StatusInternalServerError, "rebuild_error", "could not start mapping rebuild")
                return
        }
        if !started {
                writeError(w, http.StatusConflict, "already_running", "a mapping rebuild is already running")
                return
        }
        st, _ := h.mappingReb.Status(context.Background())
        writeJSON(w, http.StatusAccepted, st)
}

func nowUnix() int64 { return time.Now().Unix() }

// UpdateSettings replaces the workspace settings. Admin-only.
func (h *AdminHandler) UpdateSettings(w http.ResponseWriter, r *http.Request) {
        if !requireAdmin(w, r) {
                return
        }
        var body model.WorkspaceSettings
        if err := readJSON(r, &body); err != nil {
                writeError(w, http.StatusBadRequest, "invalid_body", err.Error())
                return
        }
        out, err := h.settings.Update(r.Context(), &body)
        if err != nil {
                writeError(w, http.StatusBadRequest, "update_error", err.Error())
                return
        }
        writeJSON(w, http.StatusOK, settingsResponse{
                MaxUploadBytes:    out.MaxUploadBytes,
                AllowedExtensions: out.AllowedExtensions,
                GiphyAPIKey:       out.GiphyAPIKey,
                GiphyEnabled:      out.GiphyAPIKey != "",
        })
}

1	package handler
2
3	import (
4	"context"
5	"net/http"
6	"time"
7
8	"github.com/DigitalTolk/ex/internal/middleware"
9	"github.com/DigitalTolk/ex/internal/model"
10	"github.com/DigitalTolk/ex/internal/search"
11	"github.com/DigitalTolk/ex/internal/service"
12	)
13
14	// SearchStatusReporter is the slim view AdminHandler needs to render
15	// the search status panel: cluster health + per-index docs/size. The
16	// concrete *search.Client satisfies this directly.
17	type SearchStatusReporter interface {
18	ClusterHealth(ctx context.Context) (map[string]any, error)
19	IndexStats(ctx context.Context) ([]search.IndexStat, error)
20	}
21
22	// MappingRebuildController is the slim view AdminHandler needs to trigger and
23	// report the cluster-coordinated users/channels mapping rebuild. The concrete
24	// *search.MappingRebuilder satisfies it; tests inject a fake.
25	type MappingRebuildController interface {
26	Start(ctx context.Context, now func() int64) (bool, error)
27	Status(ctx context.Context) (search.MappingRebuildStatus, error)
28	SchemaVersions(ctx context.Context) ([]search.SchemaVersionInfo, error)
29	}
30
31	// AdminHandler exposes admin-only endpoints for workspace configuration.
32	// Authorization is enforced inside each handler — `middleware.Auth` only
33	// confirms the caller is signed in, not that they're an admin.
34	type AdminHandler struct {
35	settings *service.SettingsService
36	searchSt SearchStatusReporter
37	reindexer *search.Reindexer
38	mappingReb MappingRebuildController
39	}
40
41	// NewAdminHandler constructs an AdminHandler.
42	func NewAdminHandler(settings service.SettingsService) AdminHandler {	33✔
43	return &AdminHandler{settings: settings}	33✔
44	}	33✔
45
46	// SetSearch wires the optional search status reporter + reindexer.
47	// Production passes the live ones; tests can pass nil and the
48	// search-admin routes return 503.
49	func (h AdminHandler) SetSearch(reporter SearchStatusReporter, reindexer search.Reindexer) {	8✔
50	h.searchSt = reporter	8✔
51	h.reindexer = reindexer	8✔
52	}	8✔
53
54	// SetMappingRebuilder wires the optional cluster-coordinated users/channels
55	// mapping rebuild (RecreateUsersChannels behind a Redis lock + shared status).
56	// nil → the mapping-rebuild route answers 503, same as an unconfigured search.
57	func (h *AdminHandler) SetMappingRebuilder(reb MappingRebuildController) {	7✔
58	h.mappingReb = reb	7✔
59	}	7✔
60
61	// settingsResponse is the wire shape returned from GetSettings. It
62	// extends model.WorkspaceSettings with a derived `giphyEnabled` flag.
63	// GIPHY requires API and media requests to be made directly by the
64	// client, so authenticated members receive the configured browser key
65	// when the picker is enabled.
66	type settingsResponse struct {
67	MaxUploadBytes int64 `json:"maxUploadBytes"`
68	AllowedExtensions []string `json:"allowedExtensions"`
69	GiphyAPIKey string `json:"giphyAPIKey,omitempty"`
70	GiphyEnabled bool `json:"giphyEnabled"`
71	}
72
73	// GetSettings returns the effective workspace settings (with defaults
74	// applied for any field the admin hasn't overridden). Available to all
75	// authenticated users so the upload UI can show the limits before
76	// attempting a request — the write side is admin-only.
77	func (h AdminHandler) GetSettings(w http.ResponseWriter, r http.Request) {	4✔
78	claims := middleware.ClaimsFromContext(r.Context())	4✔
79	if claims == nil {	5✔
80	writeError(w, http.StatusUnauthorized, "unauthorized", "authentication required")	1✔
81	return	1✔
82	}	1✔
83	ws := h.settings.Effective(r.Context())	3✔
84	resp := settingsResponse{	3✔
85	MaxUploadBytes: ws.MaxUploadBytes,	3✔
86	AllowedExtensions: ws.AllowedExtensions,	3✔
87	GiphyEnabled: ws.GiphyAPIKey != "",	3✔
88	}	3✔
89	resp.GiphyAPIKey = ws.GiphyAPIKey	3✔
90	writeJSON(w, http.StatusOK, resp)	3✔
91	}
92
93	// SearchStatus returns the search cluster's health, per-index doc
94	// counts/sizes, and the most recent reindex progress. Admin-only.
95	// Returns 503 with a structured payload (`configured: false`) when
96	// search isn't wired so the UI can render the panel without erroring.
97	func (h AdminHandler) SearchStatus(w http.ResponseWriter, r http.Request) {	7✔
98	if !requireAdmin(w, r) {	8✔
99	return	1✔
100	}	1✔
101	if h.searchSt == nil \|\| h.reindexer == nil {	7✔
102	writeJSON(w, http.StatusOK, JSON{"configured": false})	1✔
103	return	1✔
104	}	1✔
105	ctx := r.Context()	5✔
106	resp := JSON{"configured": true}	5✔
107	if health, err := h.searchSt.ClusterHealth(ctx); err == nil && health != nil {	9✔
108	resp["cluster"] = health	4✔
109	} else if err != nil {	6✔
110	resp["clusterError"] = err.Error()	1✔
111	}	1✔
112	if stats, err := h.searchSt.IndexStats(ctx); err == nil {	9✔
113	resp["indices"] = stats	4✔
114	} else {	5✔
115	resp["indicesError"] = err.Error()	1✔
116	}	1✔
117	if rst, err := h.reindexer.Status(ctx); err == nil {	10✔
118	resp["reindex"] = rst	5✔
119	} else {	5✔
NEW 120	resp["reindexError"] = err.Error()	×
NEW 121	}	×
122	if h.mappingReb != nil {	8✔
123	if st, err := h.mappingReb.Status(ctx); err == nil {	5✔
124	resp["mappingRebuild"] = st	2✔
125	} else {	3✔
126	resp["mappingRebuildError"] = err.Error()	1✔
127	}	1✔
128	if versions, err := h.mappingReb.SchemaVersions(ctx); err == nil {	5✔
129	resp["schemaVersions"] = versions	2✔
130	} else {	3✔
131	resp["schemaVersionsError"] = err.Error()	1✔
132	}	1✔
133	}
134	writeJSON(w, http.StatusOK, resp)	5✔
135	}
136
137	// StartSearchReindex kicks off a full rebuild in the background.
138	// Admin-only. Returns 202 if a fresh run started, 409 if one is
139	// already running, 503 if search isn't configured.
140	func (h AdminHandler) StartSearchReindex(w http.ResponseWriter, r http.Request) {	4✔
141	if !requireAdmin(w, r) {	5✔
142	return	1✔
143	}	1✔
144	if h.reindexer == nil {	4✔
145	writeError(w, http.StatusServiceUnavailable, "search_disabled", "search is not configured")	1✔
146	return	1✔
147	}	1✔
148	// Detach the request context so the goroutine survives the HTTP
149	// response. Reindexes routinely outlive the request timeout —
150	// admins poll status via SearchStatus afterwards.
151	if !h.reindexer.Start(context.Background()) {	3✔
152	writeError(w, http.StatusConflict, "already_running", "a reindex is already running")	1✔
153	return	1✔
154	}	1✔
155	st, err := h.reindexer.Status(r.Context())	1✔
156	if err != nil {	1✔
NEW 157	// The run started; a status read-back failure is non-fatal — report	×
NEW 158	// accepted with an empty snapshot, the panel will poll.	×
NEW 159	writeJSON(w, http.StatusAccepted, search.ReindexProgress{Running: true})	×
NEW 160	return	×
NEW 161	}	×
162	writeJSON(w, http.StatusAccepted, st)	1✔
163	}
164
165	// StartSearchMappingRebuild kicks off the cluster-coordinated users/channels
166	// mapping rebuild (staging index → atomic alias-swap) — the path that actually
167	// rolls a new analyzer onto an existing cluster. Admin-only. A Redis lock makes
168	// it single-flight across every instance: 202 if THIS call won the lock and
169	// started a run, 409 if another instance (or a still-cooling crashed run) holds
170	// it, 503 if search isn't configured, 500 on a coordination (Redis) error.
171	func (h AdminHandler) StartSearchMappingRebuild(w http.ResponseWriter, r http.Request) {	5✔
172	if !requireAdmin(w, r) {	6✔
173	return	1✔
174	}	1✔
175	if h.mappingReb == nil {	5✔
176	writeError(w, http.StatusServiceUnavailable, "search_disabled", "search is not configured")	1✔
177	return	1✔
178	}	1✔
179	// Detach from the request context so the lock/state writes and the detached
180	// rebuild goroutine survive the HTTP response; admins poll status afterwards.
181	started, err := h.mappingReb.Start(context.Background(), nowUnix)	3✔
182	if err != nil {	4✔
183	// Generic 500 — don't leak the Redis/coordination error to the client.	1✔
184	writeError(w, http.StatusInternalServerError, "rebuild_error", "could not start mapping rebuild")	1✔
185	return	1✔
186	}	1✔
187	if !started {	3✔
188	writeError(w, http.StatusConflict, "already_running", "a mapping rebuild is already running")	1✔
189	return	1✔
190	}	1✔
191	st, _ := h.mappingReb.Status(context.Background())	1✔
192	writeJSON(w, http.StatusAccepted, st)	1✔
193	}
194
195	func nowUnix() int64 { return time.Now().Unix() }	1✔
196
197	// UpdateSettings replaces the workspace settings. Admin-only.
198	func (h AdminHandler) UpdateSettings(w http.ResponseWriter, r http.Request) {	5✔
199	if !requireAdmin(w, r) {	6✔
200	return	1✔
201	}	1✔
202	var body model.WorkspaceSettings	4✔
203	if err := readJSON(r, &body); err != nil {	5✔
204	writeError(w, http.StatusBadRequest, "invalid_body", err.Error())	1✔
205	return	1✔
206	}	1✔
207	out, err := h.settings.Update(r.Context(), &body)	3✔
208	if err != nil {	4✔
209	writeError(w, http.StatusBadRequest, "update_error", err.Error())	1✔
210	return	1✔
211	}	1✔
212	writeJSON(w, http.StatusOK, settingsResponse{	2✔
213	MaxUploadBytes: out.MaxUploadBytes,	2✔
214	AllowedExtensions: out.AllowedExtensions,	2✔
215	GiphyAPIKey: out.GiphyAPIKey,	2✔
216	GiphyEnabled: out.GiphyAPIKey != "",	2✔
217	})	2✔
218	}

DigitalTolk / ex / 28705447299

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous