mirror of
https://github.com/hanzo-fi/payments.git
synced 2026-10-30 20:31:20 +00:00
CrimsonandGitHub
db0561d07e
* Set webhook ID as payment adjustment ID when adjustment created via webhook * Fetch Column payments oldest to newest * Ensure that webhook secrets are fetched when containers restart * Regenerate plugin code for tests * Add paging to fetch webhook request * Add WorkerPool unit test to increase test coverage * Guard against unhandled webhooks in Column plugin initiation * Rename Timeline.LatestID to Timeline.LastSeenID for clarity * Add metadata column to webhooks_configs table * Persist webhook_configs.metadata * Persist Column webhook secrets as metadata in webhooks_configs * Remove PSPWebhookConfig from InstallResponse * Stop passing context to initWebhookConfig() * Don't pass context to RegisterPlugin func * Don't pass context to plugins.RegisterPlugin
98 lines
2.4 KiB
Go
98 lines
2.4 KiB
Go
package workflow
|
|
|
|
import (
|
|
"fmt"
|
|
"net/url"
|
|
|
|
"github.com/formancehq/payments/internal/connectors/engine/activities"
|
|
"github.com/formancehq/payments/internal/models"
|
|
"go.temporal.io/api/enums/v1"
|
|
"go.temporal.io/sdk/workflow"
|
|
)
|
|
|
|
type CreateWebhooks struct {
|
|
ConnectorID models.ConnectorID
|
|
Config models.Config
|
|
FromPayload *FromPayload
|
|
}
|
|
|
|
func (w Workflow) runCreateWebhooks(
|
|
ctx workflow.Context,
|
|
createWebhooks CreateWebhooks,
|
|
nextTasks []models.ConnectorTaskTree,
|
|
) error {
|
|
return w.createWebhooks(ctx, createWebhooks, nextTasks)
|
|
}
|
|
|
|
func (w Workflow) createWebhooks(
|
|
ctx workflow.Context,
|
|
createWebhooks CreateWebhooks,
|
|
nextTasks []models.ConnectorTaskTree,
|
|
) error {
|
|
webhookBaseURL, err := url.JoinPath(w.stackPublicURL, "api/payments/v3/connectors/webhooks", createWebhooks.ConnectorID.String())
|
|
if err != nil {
|
|
return fmt.Errorf("joining webhook base URL: %w", err)
|
|
}
|
|
|
|
resp, err := activities.PluginCreateWebhooks(
|
|
infiniteRetryContext(ctx),
|
|
createWebhooks.ConnectorID,
|
|
models.CreateWebhooksRequest{
|
|
WebhookBaseUrl: webhookBaseURL,
|
|
ConnectorID: createWebhooks.ConnectorID.String(),
|
|
FromPayload: createWebhooks.FromPayload.GetPayload(),
|
|
},
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to create webhooks: %w", err)
|
|
}
|
|
|
|
if len(resp.Configs) > 0 {
|
|
configs := make([]models.WebhookConfig, 0, len(resp.Configs))
|
|
for _, c := range resp.Configs {
|
|
configs = append(configs, models.WebhookConfig{
|
|
Name: c.Name,
|
|
ConnectorID: createWebhooks.ConnectorID,
|
|
URLPath: c.URLPath,
|
|
Metadata: c.Metadata,
|
|
})
|
|
}
|
|
err = activities.StorageWebhooksConfigsStore(
|
|
infiniteRetryContext(ctx),
|
|
configs,
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("storing webhooks: %w", err)
|
|
}
|
|
}
|
|
|
|
for _, other := range resp.Others {
|
|
if err := workflow.ExecuteChildWorkflow(
|
|
workflow.WithChildOptions(
|
|
ctx,
|
|
workflow.ChildWorkflowOptions{
|
|
TaskQueue: w.getDefaultTaskQueue(),
|
|
ParentClosePolicy: enums.PARENT_CLOSE_POLICY_ABANDON,
|
|
SearchAttributes: map[string]interface{}{
|
|
SearchAttributeStack: w.stack,
|
|
},
|
|
},
|
|
),
|
|
Run,
|
|
createWebhooks.Config,
|
|
createWebhooks.ConnectorID,
|
|
&FromPayload{
|
|
ID: other.ID,
|
|
Payload: other.Other,
|
|
},
|
|
nextTasks,
|
|
).Get(ctx, nil); err != nil {
|
|
return fmt.Errorf("running next workflow: %w", err)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
const RunCreateWebhooks = "RunCreateWebhooks"
|