package 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 }