Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
8.6 kB · 271 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272package spxrpc
import ( "context" "encoding/json" "net/http" "net/url" "strconv" "strings"
"github.com/labstack/echo/v4" "github.com/streamplace/oatproxy/pkg/oatproxy" "stream.place/streamplace/pkg/log" placestream "stream.place/streamplace/pkg/placestream" "stream.place/streamplace/pkg/statedb")
func (s *Server) handlePlaceStreamServerCreateWebhook(ctx context.Context, input *placestream.ServerCreateWebhook_Input) (*placestream.ServerCreateWebhook_Output, error) { // Get authenticated user session, _ := oatproxy.GetOAuthSession(ctx) if session == nil { return nil, echo.NewHTTPError(http.StatusUnauthorized, "oauth session not found") }
// Validate input if input.Url == "" { return nil, echo.NewHTTPError(http.StatusBadRequest, "URL is required") } if len(input.Events) == 0 { return nil, echo.NewHTTPError(http.StatusBadRequest, "At least one event type is required") }
// Validate URL format if _, err := url.Parse(input.Url); err != nil { return nil, echo.NewHTTPError(http.StatusBadRequest, "Invalid URL format") } // Convert input to database model using the conversion function webhook, err := statedb.WebhookFromLexiconInput(*input, session.DID, "") // ID will be generated by the database if err != nil { log.Error(ctx, "failed to convert input to webhook", "err", err) return nil, echo.NewHTTPError(http.StatusInternalServerError, "Failed to create webhook") }
// Create webhook err = s.statefulDB.CreateWebhook(webhook) if err != nil { log.Error(ctx, "failed to create webhook in database", "err", err, "url", input.Url, "events", input.Events) return nil, echo.NewHTTPError(http.StatusInternalServerError, "Failed to create webhook") }
// Convert to API response apiWebhook, err := webhook.ToLexicon() if err != nil { log.Error(ctx, "failed to convert webhook to API format", "err", err) return nil, echo.NewHTTPError(http.StatusInternalServerError, "Failed to format webhook response") }
return &placestream.ServerCreateWebhook_Output{ Webhook: apiWebhook, }, nil}
func (s *Server) handlePlaceStreamServerListWebhooks(ctx context.Context, active bool, cursor string, event string, limit int) (*placestream.ServerListWebhooks_Output, error) { // Get authenticated user session, _ := oatproxy.GetOAuthSession(ctx) if session == nil { return nil, echo.NewHTTPError(http.StatusUnauthorized, "oauth session not found") }
// Set default limit if limit <= 0 || limit > 100 { limit = 50 }
// Parse cursor for offset offset := 0 if cursor != "" { var err error offset, err = strconv.Atoi(cursor) if err != nil { return nil, echo.NewHTTPError(http.StatusBadRequest, "Invalid cursor") } }
// Build filters. The generated stub can't distinguish an absent `active` // param from an explicit active=false (both arrive as false), so filtering // only applies when active=true. Omitting the param or passing // active=false returns all webhooks regardless of status. filters := make(map[string]interface{}) if active { filters["active"] = active }
// Get webhooks webhooks, err := s.statefulDB.ListWebhooks(session.DID, limit+1, offset, filters) if err != nil { log.Error(ctx, "failed to list webhooks", "err", err) return nil, echo.NewHTTPError(http.StatusInternalServerError, "Failed to list webhooks") }
// Filter by event type if specified if event != "" { filtered := make([]statedb.Webhook, 0) for _, w := range webhooks { var events []string if err := json.Unmarshal(w.Events, &events); err == nil { for _, e := range events { if e == event { filtered = append(filtered, w) break } } } } webhooks = filtered }
// Check if there are more results var nextCursor *string if len(webhooks) > limit { webhooks = webhooks[:limit] next := strconv.Itoa(offset + limit) nextCursor = &next }
// Convert to API format apiWebhooks := make([]placestream.ServerDefs_Webhook, len(webhooks)) for i, webhook := range webhooks { apiWebhook, err := webhook.ToLexicon() if err != nil { log.Error(ctx, "failed to convert webhook to API format", "err", err) continue } apiWebhooks[i] = apiWebhook }
return &placestream.ServerListWebhooks_Output{ Webhooks: apiWebhooks, Cursor: nextCursor, }, nil}
func (s *Server) handlePlaceStreamServerGetWebhook(ctx context.Context, id string) (*placestream.ServerGetWebhook_Output, error) { // Get authenticated user session, _ := oatproxy.GetOAuthSession(ctx) if session == nil { return nil, echo.NewHTTPError(http.StatusUnauthorized, "oauth session not found") }
// Get webhook webhook, err := s.statefulDB.GetWebhook(id, session.DID) if err != nil { if strings.Contains(err.Error(), "record not found") { return nil, echo.NewHTTPError(http.StatusNotFound, "Webhook not found") } log.Error(ctx, "failed to get webhook", "err", err) return nil, echo.NewHTTPError(http.StatusInternalServerError, "Failed to get webhook") }
// Convert to API format apiWebhook, err := webhook.ToLexicon() if err != nil { log.Error(ctx, "failed to convert webhook to API format", "err", err) return nil, echo.NewHTTPError(http.StatusInternalServerError, "Failed to format webhook response") }
return &placestream.ServerGetWebhook_Output{ Webhook: apiWebhook, }, nil}
func (s *Server) handlePlaceStreamServerUpdateWebhook(ctx context.Context, input *placestream.ServerUpdateWebhook_Input) (*placestream.ServerUpdateWebhook_Output, error) { // Get authenticated user session, _ := oatproxy.GetOAuthSession(ctx) if session == nil { return nil, echo.NewHTTPError(http.StatusUnauthorized, "oauth session not found") }
// Validate URL if provided if input.Url != nil { if _, err := url.Parse(*input.Url); err != nil { return nil, echo.NewHTTPError(http.StatusBadRequest, "Invalid URL format") } }
// Build updates map updates := make(map[string]interface{}) if input.Url != nil { updates["url"] = *input.Url } if input.Events != nil { eventsJSON, err := json.Marshal(input.Events) if err != nil { return nil, echo.NewHTTPError(http.StatusBadRequest, "Invalid events format") } updates["events"] = json.RawMessage(eventsJSON) } if input.Active != nil { updates["active"] = *input.Active } if input.Prefix != nil { updates["prefix"] = *input.Prefix } if input.Suffix != nil { updates["suffix"] = *input.Suffix } if input.Rewrite != nil { rewriteJSON, err := json.Marshal(input.Rewrite) if err != nil { return nil, echo.NewHTTPError(http.StatusBadRequest, "Invalid rewrite rules format") } updates["rewrite"] = json.RawMessage(rewriteJSON) } if input.Name != nil { updates["name"] = *input.Name } if input.Description != nil { updates["description"] = *input.Description } if input.MuteWords != nil { muteWordsJSON, err := json.Marshal(input.MuteWords) if err != nil { return nil, echo.NewHTTPError(http.StatusBadRequest, "Invalid mute words format") } updates["mute_words"] = json.RawMessage(muteWordsJSON) }
if len(updates) == 0 { return nil, echo.NewHTTPError(http.StatusBadRequest, "No fields to update") }
// Update webhook webhook, err := s.statefulDB.UpdateWebhook(input.Id, session.DID, updates) if err != nil { if strings.Contains(err.Error(), "record not found") { return nil, echo.NewHTTPError(http.StatusNotFound, "Webhook not found") } log.Error(ctx, "failed to update webhook", "err", err) return nil, echo.NewHTTPError(http.StatusInternalServerError, "Failed to update webhook") }
// Convert to API format apiWebhook, err := webhook.ToLexicon() if err != nil { log.Error(ctx, "failed to convert webhook to API format", "err", err) return nil, echo.NewHTTPError(http.StatusInternalServerError, "Failed to format webhook response") }
return &placestream.ServerUpdateWebhook_Output{ Webhook: apiWebhook, }, nil}
func (s *Server) handlePlaceStreamServerDeleteWebhook(ctx context.Context, input *placestream.ServerDeleteWebhook_Input) (*placestream.ServerDeleteWebhook_Output, error) { // Get authenticated user session, _ := oatproxy.GetOAuthSession(ctx) if session == nil { return nil, echo.NewHTTPError(http.StatusUnauthorized, "oauth session not found") }
// Delete webhook err := s.statefulDB.DeleteWebhook(input.Id, session.DID) if err != nil { log.Error(ctx, "failed to delete webhook", "err", err) return nil, echo.NewHTTPError(http.StatusInternalServerError, "Failed to delete webhook") }
return &placestream.ServerDeleteWebhook_Output{ Success: true, }, nil}