18894867512

Committed 29 Oct 2025 02:09AM UTC coverage: 37.809% (-0.03%) from 37.841%

Build # 18894867512

Build Type

Pull #9

github

Committed by

rom8726

Commit Message

DLQ API plugin implemented.

Pull Request Pull Request #9: Dead Letter Queue implemented

Run Details

245 of 685 new or added lines in 6 files covered. (35.77%)

6 existing lines in 1 file now uncovered.

2861 of 7567 relevant lines covered (37.81%)

32.25 hits per line

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

90.99

/plugins/api/dlq/plugin.go

package dlq

import (
        "encoding/json"
        "errors"
        "io"
        "net/http"
        "strconv"
        "time"

        "github.com/rom8726/floxy"
        "github.com/rom8726/floxy/api"
)

var _ api.Plugin = (*Plugin)(nil)

type Plugin struct {
        engine floxy.IEngine
        store  floxy.Store
}

func New(engine floxy.IEngine, store floxy.Store) *Plugin {
        return &Plugin{engine: engine, store: store}
}

func (p *Plugin) Name() string        { return "dlq" }
func (p *Plugin) Description() string { return "Dead Letter Queue operations" }

func (p *Plugin) RegisterRoutes(mux *http.ServeMux) {
        mux.HandleFunc("GET /api/dlq", HandleList(p.store))
        mux.HandleFunc("GET /api/dlq/{id}", HandleGet(p.store))
        mux.HandleFunc("POST /api/dlq/{id}/requeue", HandleRequeue(p.engine))
}

func HandleList(store floxy.Store) func(w http.ResponseWriter, r *http.Request) {
        return func(w http.ResponseWriter, r *http.Request) {
                ctx := r.Context()

                // pagination: default page=1, page_size=20
                page := 1
                pageSize := 20
                if v := r.URL.Query().Get("page"); v != "" {
                        if n, err := strconv.Atoi(v); err == nil && n > 0 {
                                page = n
                        }
                }
                if v := r.URL.Query().Get("page_size"); v != "" {
                        if n, err := strconv.Atoi(v); err == nil && n > 0 && n <= 1000 {
                                pageSize = n
                        }
                }

                offset := (page - 1) * pageSize
                items, total, err := store.ListDeadLetters(ctx, offset, pageSize)
                if err != nil {
                        api.WriteErrorResponse(w, err, http.StatusInternalServerError)
                        return
                }

                resp := ListResponse{
                        Items:    make([]DeadLetterResponse, 0, len(items)),
                        Page:     page,
                        PageSize: pageSize,
                        Total:    total,
                }
                for _, it := range items {
                        resp.Items = append(resp.Items, DeadLetterResponse{
                                ID:         it.ID,
                                InstanceID: it.InstanceID,
                                WorkflowID: it.WorkflowID,
                                StepID:     it.StepID,
                                StepName:   it.StepName,
                                StepType:   it.StepType,
                                Input:      it.Input,
                                Error:      it.Error,
                                Reason:     it.Reason,
                                CreatedAt:  it.CreatedAt.UTC().Format(time.RFC3339Nano),
                        })
                }

                w.Header().Set("Content-Type", "application/json")
                w.WriteHeader(http.StatusOK)
                _ = json.NewEncoder(w).Encode(resp)
        }
}

func HandleGet(store floxy.Store) func(w http.ResponseWriter, r *http.Request) {
        return func(w http.ResponseWriter, r *http.Request) {
                ctx := r.Context()

                idStr := r.PathValue("id")
                id, err := strconv.ParseInt(idStr, 10, 64)
                if err != nil {
                        api.WriteErrorResponse(w, err, http.StatusBadRequest)
                        return
                }

                rec, err := store.GetDeadLetterByID(ctx, id)
                if err != nil {
                        if errors.Is(err, floxy.ErrEntityNotFound) {
                                api.WriteErrorResponse(w, err, http.StatusNotFound)
                                return
                        }
                        api.WriteErrorResponse(w, err, http.StatusInternalServerError)
                        return
                }

                resp := DeadLetterResponse{
                        ID:         rec.ID,
                        InstanceID: rec.InstanceID,
                        WorkflowID: rec.WorkflowID,
                        StepID:     rec.StepID,
                        StepName:   rec.StepName,
                        StepType:   rec.StepType,
                        Input:      rec.Input,
                        Error:      rec.Error,
                        Reason:     rec.Reason,
                        CreatedAt:  rec.CreatedAt.UTC().Format(time.RFC3339Nano),
                }

                w.Header().Set("Content-Type", "application/json")
                w.WriteHeader(http.StatusOK)
                _ = json.NewEncoder(w).Encode(resp)
        }
}

func HandleRequeue(engine floxy.IEngine) func(w http.ResponseWriter, r *http.Request) {
        return func(w http.ResponseWriter, r *http.Request) {
                ctx := r.Context()

                idStr := r.PathValue("id")
                id, err := strconv.ParseInt(idStr, 10, 64)
                if err != nil {
                        api.WriteErrorResponse(w, err, http.StatusBadRequest)
                        return
                }

                var req RequeueRequest
                if err := json.NewDecoder(r.Body).Decode(&req); err != nil && !errors.Is(err, io.EOF) {
                        api.WriteErrorResponse(w, err, http.StatusBadRequest)
                        return
                }

                if err := engine.RequeueFromDLQ(ctx, id, req.NewInput); err != nil {
                        if errors.Is(err, floxy.ErrEntityNotFound) {
                                api.WriteErrorResponse(w, err, http.StatusNotFound)
                                return
                        }
                        api.WriteErrorResponse(w, err, http.StatusInternalServerError)
                        return
                }

                w.WriteHeader(http.StatusNoContent)
        }
}

1	package dlq
2
3	import (
4	"encoding/json"
5	"errors"
6	"io"
7	"net/http"
8	"strconv"
9	"time"
10
11	"github.com/rom8726/floxy"
12	"github.com/rom8726/floxy/api"
13	)
14
15	var _ api.Plugin = (*Plugin)(nil)
16
17	type Plugin struct {
18	engine floxy.IEngine
19	store floxy.Store
20	}
21
NEW 22	func New(engine floxy.IEngine, store floxy.Store) *Plugin {	×
NEW 23	return &Plugin{engine: engine, store: store}	×
NEW 24	}	×
25
NEW 26	func (p *Plugin) Name() string { return "dlq" }	×
NEW 27	func (p *Plugin) Description() string { return "Dead Letter Queue operations" }	×
28
NEW 29	func (p Plugin) RegisterRoutes(mux http.ServeMux) {	×
NEW 30	mux.HandleFunc("GET /api/dlq", HandleList(p.store))	×
NEW 31	mux.HandleFunc("GET /api/dlq/{id}", HandleGet(p.store))	×
NEW 32	mux.HandleFunc("POST /api/dlq/{id}/requeue", HandleRequeue(p.engine))	×
NEW 33	}	×
34
35	func HandleList(store floxy.Store) func(w http.ResponseWriter, r *http.Request) {	3✔
36	return func(w http.ResponseWriter, r *http.Request) {	6✔
37	ctx := r.Context()	3✔
38		3✔
39	// pagination: default page=1, page_size=20	3✔
40	page := 1	3✔
41	pageSize := 20	3✔
42	if v := r.URL.Query().Get("page"); v != "" {	4✔
43	if n, err := strconv.Atoi(v); err == nil && n > 0 {	2✔
44	page = n	1✔
45	}	1✔
46	}
47	if v := r.URL.Query().Get("page_size"); v != "" {	4✔
48	if n, err := strconv.Atoi(v); err == nil && n > 0 && n <= 1000 {	2✔
49	pageSize = n	1✔
50	}	1✔
51	}
52
53	offset := (page - 1) * pageSize	3✔
54	items, total, err := store.ListDeadLetters(ctx, offset, pageSize)	3✔
55	if err != nil {	4✔
56	api.WriteErrorResponse(w, err, http.StatusInternalServerError)	1✔
57	return	1✔
58	}	1✔
59
60	resp := ListResponse{	2✔
61	Items: make([]DeadLetterResponse, 0, len(items)),	2✔
62	Page: page,	2✔
63	PageSize: pageSize,	2✔
64	Total: total,	2✔
65	}	2✔
66	for _, it := range items {	3✔
67	resp.Items = append(resp.Items, DeadLetterResponse{	1✔
68	ID: it.ID,	1✔
69	InstanceID: it.InstanceID,	1✔
70	WorkflowID: it.WorkflowID,	1✔
71	StepID: it.StepID,	1✔
72	StepName: it.StepName,	1✔
73	StepType: it.StepType,	1✔
74	Input: it.Input,	1✔
75	Error: it.Error,	1✔
76	Reason: it.Reason,	1✔
77	CreatedAt: it.CreatedAt.UTC().Format(time.RFC3339Nano),	1✔
78	})	1✔
79	}	1✔
80
81	w.Header().Set("Content-Type", "application/json")	2✔
82	w.WriteHeader(http.StatusOK)	2✔
83	_ = json.NewEncoder(w).Encode(resp)	2✔
84	}
85	}
86
87	func HandleGet(store floxy.Store) func(w http.ResponseWriter, r *http.Request) {	4✔
88	return func(w http.ResponseWriter, r *http.Request) {	8✔
89	ctx := r.Context()	4✔
90		4✔
91	idStr := r.PathValue("id")	4✔
92	id, err := strconv.ParseInt(idStr, 10, 64)	4✔
93	if err != nil {	5✔
94	api.WriteErrorResponse(w, err, http.StatusBadRequest)	1✔
95	return	1✔
96	}	1✔
97
98	rec, err := store.GetDeadLetterByID(ctx, id)	3✔
99	if err != nil {	5✔
100	if errors.Is(err, floxy.ErrEntityNotFound) {	3✔
101	api.WriteErrorResponse(w, err, http.StatusNotFound)	1✔
102	return	1✔
103	}	1✔
104	api.WriteErrorResponse(w, err, http.StatusInternalServerError)	1✔
105	return	1✔
106	}
107
108	resp := DeadLetterResponse{	1✔
109	ID: rec.ID,	1✔
110	InstanceID: rec.InstanceID,	1✔
111	WorkflowID: rec.WorkflowID,	1✔
112	StepID: rec.StepID,	1✔
113	StepName: rec.StepName,	1✔
114	StepType: rec.StepType,	1✔
115	Input: rec.Input,	1✔
116	Error: rec.Error,	1✔
117	Reason: rec.Reason,	1✔
118	CreatedAt: rec.CreatedAt.UTC().Format(time.RFC3339Nano),	1✔
119	}	1✔
120		1✔
121	w.Header().Set("Content-Type", "application/json")	1✔
122	w.WriteHeader(http.StatusOK)	1✔
123	_ = json.NewEncoder(w).Encode(resp)	1✔
124	}
125	}
126
127	func HandleRequeue(engine floxy.IEngine) func(w http.ResponseWriter, r *http.Request) {	6✔
128	return func(w http.ResponseWriter, r *http.Request) {	12✔
129	ctx := r.Context()	6✔
130		6✔
131	idStr := r.PathValue("id")	6✔
132	id, err := strconv.ParseInt(idStr, 10, 64)	6✔
133	if err != nil {	7✔
134	api.WriteErrorResponse(w, err, http.StatusBadRequest)	1✔
135	return	1✔
136	}	1✔
137
138	var req RequeueRequest	5✔
139	if err := json.NewDecoder(r.Body).Decode(&req); err != nil && !errors.Is(err, io.EOF) {	6✔
140	api.WriteErrorResponse(w, err, http.StatusBadRequest)	1✔
141	return	1✔
142	}	1✔
143
144	if err := engine.RequeueFromDLQ(ctx, id, req.NewInput); err != nil {	6✔
145	if errors.Is(err, floxy.ErrEntityNotFound) {	3✔
146	api.WriteErrorResponse(w, err, http.StatusNotFound)	1✔
147	return	1✔
148	}	1✔
149	api.WriteErrorResponse(w, err, http.StatusInternalServerError)	1✔
150	return	1✔
151	}
152
153	w.WriteHeader(http.StatusNoContent)	2✔
154	}
155	}

rom8726 / floxy / 18894867512

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