feat: wise.com connector (#23)

* add structure for wise connector

* fix: first

* fix: lint

* fix: GET using id

* chore: first

* chore: bump go-libs

* chore: bump go-libs

* chore: add dependabot

* chore: move message models

* chore: new method to create kafka message

* chore: move kafka messages to core pkg

* chore: update messages

* chore: revert moving files

* chore: update messages

* chore(deps): bump github/codeql-action from 1 to 2

Bumps [github/codeql-action](https://github.com/github/codeql-action) from 1 to 2.
- [Release notes](https://github.com/github/codeql-action/releases)
- [Changelog](https://github.com/github/codeql-action/blob/main/CHANGELOG.md)
- [Commits](github/codeql-action@v1...v2)

---
updated-dependencies:
- dependency-name: github/codeql-action
  dependency-type: direct:production
  update-type: version-update:semver-major
...

Signed-off-by: dependabot[bot] <support@github.com>

* fix: bad dependencies

* fix: payments publish

* chore(deps): bump github.com/numary/go-libs from 1.0.0 to 1.0.1

Bumps [github.com/numary/go-libs](https://github.com/numary/go-libs) from 1.0.0 to 1.0.1.
- [Release notes](https://github.com/numary/go-libs/releases)
- [Commits](formancehq/go-libs@v1.0.0...v1.0.1)

---
updated-dependencies:
- dependency-name: github.com/numary/go-libs
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>

* fix: Ingestion message payload marshaling (#32)

* chore(deps): bump docker/build-push-action from 2 to 3

Bumps [docker/build-push-action](https://github.com/docker/build-push-action) from 2 to 3.
- [Release notes](https://github.com/docker/build-push-action/releases)
- [Commits](docker/build-push-action@v2...v3)

---
updated-dependencies:
- dependency-name: docker/build-push-action
  dependency-type: direct:production
  update-type: version-update:semver-major
...

Signed-off-by: dependabot[bot] <support@github.com>

* chore(deps): bump docker/setup-qemu-action from 1 to 2

Bumps [docker/setup-qemu-action](https://github.com/docker/setup-qemu-action) from 1 to 2.
- [Release notes](https://github.com/docker/setup-qemu-action/releases)
- [Commits](docker/setup-qemu-action@v1...v2)

---
updated-dependencies:
- dependency-name: docker/setup-qemu-action
  dependency-type: direct:production
  update-type: version-update:semver-major
...

Signed-off-by: dependabot[bot] <support@github.com>

* chore(deps): bump docker/login-action from 1 to 2

Bumps [docker/login-action](https://github.com/docker/login-action) from 1 to 2.
- [Release notes](https://github.com/docker/login-action/releases)
- [Commits](docker/login-action@v1...v2)

---
updated-dependencies:
- dependency-name: docker/login-action
  dependency-type: direct:production
  update-type: version-update:semver-major
...

Signed-off-by: dependabot[bot] <support@github.com>

* chore(deps): bump actions/checkout from 2 to 3

Bumps [actions/checkout](https://github.com/actions/checkout) from 2 to 3.
- [Release notes](https://github.com/actions/checkout/releases)
- [Changelog](https://github.com/actions/checkout/blob/main/CHANGELOG.md)
- [Commits](actions/checkout@v2...v3)

---
updated-dependencies:
- dependency-name: actions/checkout
  dependency-type: direct:production
  update-type: version-update:semver-major
...

Signed-off-by: dependabot[bot] <support@github.com>

* chore(deps): bump docker/setup-buildx-action from 1 to 2

Bumps [docker/setup-buildx-action](https://github.com/docker/setup-buildx-action) from 1 to 2.
- [Release notes](https://github.com/docker/setup-buildx-action/releases)
- [Commits](docker/setup-buildx-action@v1...v2)

---
updated-dependencies:
- dependency-name: docker/setup-buildx-action
  dependency-type: direct:production
  update-type: version-update:semver-major
...

Signed-off-by: dependabot[bot] <support@github.com>

* feat: dummypay connector

* feat: dummypay connector

Signed-off-by: Lawrence Zawila <113581282+darkmatterpool@users.noreply.github.com>

* feat: uncommented dummypay load

Signed-off-by: Lawrence Zawila <113581282+darkmatterpool@users.noreply.github.com>

* feat: dummypay payments generation

Signed-off-by: Lawrence Zawila <113581282+darkmatterpool@users.noreply.github.com>

* feat: dummypay connector tests

Signed-off-by: Lawrence Zawila <113581282+darkmatterpool@users.noreply.github.com>

* fix: linter fixes

Signed-off-by: Lawrence Zawila <113581282+darkmatterpool@users.noreply.github.com>

* ci: Switch to FormanceHQ Organization

* fix: drop syscall as it's not supported on windows arm

Signed-off-by: Lawrence Zawila <113581282+darkmatterpool@users.noreply.github.com>

* feat: add scheme generation

Signed-off-by: Lawrence Zawila <113581282+darkmatterpool@users.noreply.github.com>

* fix: typo in error message

Signed-off-by: Lawrence Zawila <113581282+darkmatterpool@users.noreply.github.com>

* feat: add swagger examples for dummypay

Signed-off-by: Lawrence Zawila <113581282+darkmatterpool@users.noreply.github.com>

* add structure for wise connector

* fix: general refactoring

Signed-off-by: Lawrence Zawila <113581282+darkmatterpool@users.noreply.github.com>

* feat: complete wise integration

Signed-off-by: Lawrence Zawila <113581282+darkmatterpool@users.noreply.github.com>

* feat: add payment status mapping

Signed-off-by: Lawrence Zawila <113581282+darkmatterpool@users.noreply.github.com>

* feat: add additional error checks

Signed-off-by: Lawrence Zawila <113581282+darkmatterpool@users.noreply.github.com>

Signed-off-by: dependabot[bot] <support@github.com>
Signed-off-by: Lawrence Zawila <113581282+darkmatterpool@users.noreply.github.com>
Co-authored-by: Antoine Gelloz <antoine.gelloz@me.com>
Co-authored-by: Maxence Maireaux <maxence@maireaux.fr>
Co-authored-by: Geoffrey Ragot <geoffrey.ragot@gmail.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: Lawrence Zawila <113581282+darkmatterpool@users.noreply.github.com>

cmd/server.go

@@ -24,6 +24,7 @@ import (
 	"github.com/numary/payments/pkg/bridge/cdi"
 	"github.com/numary/payments/pkg/bridge/connectors/dummypay"
 	"github.com/numary/payments/pkg/bridge/connectors/stripe"
+	"github.com/numary/payments/pkg/bridge/connectors/wise"
 	bridgeHttp "github.com/numary/payments/pkg/bridge/http"
 	"github.com/numary/payments/pkg/database"
 	paymentapi "github.com/numary/payments/pkg/http"
@@ -283,5 +284,9 @@ func HTTPModule() fx.Option {
 			viper.GetBool(authBearerUseScopesFlag),
 			dummypay.NewLoader(),
 		),
+		cdi.ConnectorModule(
+			viper.GetBool(authBearerUseScopesFlag),
+			wise.NewLoader(),
+		),
 	)
 }

pkg/bridge/connectors/wise/client.go

@@ -0,0 +1,132 @@
+package wise
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+)
+
+const apiEndpoint = "https://api.wise.com"
+
+type apiTransport struct {
+	ApiKey string
+}
+
+func (t *apiTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", t.ApiKey))
+	return http.DefaultTransport.RoundTrip(req)
+}
+
+type client struct {
+	httpClient *http.Client
+}
+
+type profile struct {
+	ID   uint64 `json:"id"`
+	Type string `json:"type"`
+}
+
+type transfer struct {
+	ID                    uint64  `json:"id"`
+	Reference             string  `json:"reference"`
+	Status                string  `json:"status"`
+	SourceAccount         uint64  `json:"sourceAccount"`
+	SourceCurrency        string  `json:"sourceCurrency"`
+	SourceValue           float64 `json:"sourceValue"`
+	TargetAccount         uint64  `json:"targetAccount"`
+	TargetCurrency        string  `json:"targetCurrency"`
+	TargetValue           float64 `json:"targetValue"`
+	Business              uint64  `json:"business"`
+	Created               string  `json:"created"`
+	CustomerTransactionId string  `json:"customerTransactionId"`
+	Details               struct {
+		Reference string `json:"reference"`
+	} `json:"details"`
+	Rate float64 `json:"rate"`
+	User uint64  `json:"user"`
+}
+
+func (w *client) endpoint(path string) string {
+	return fmt.Sprintf("%s/%s", apiEndpoint, path)
+}
+
+func (w *client) getProfiles() ([]profile, error) {
+	var profiles []profile
+
+	res, err := w.httpClient.Get(w.endpoint("v1/profiles"))
+	if err != nil {
+		return profiles, err
+	}
+
+	b, err := io.ReadAll(res.Body)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read response body: %w", err)
+	}
+
+	err = json.Unmarshal(b, &profiles)
+	if err != nil {
+		return nil, fmt.Errorf("failed to unmarshal profiles: %w", err)
+	}
+
+	return profiles, nil
+}
+
+func (w *client) getTransfers(profile *profile) ([]transfer, error) {
+	var transfers []transfer
+
+	limit := 10
+	offset := 0
+
+	for {
+		var ts []transfer
+
+		req, err := http.NewRequest(http.MethodGet, w.endpoint("v1/transfers"), nil)
+		if err != nil {
+			return transfers, err
+		}
+
+		q := req.URL.Query()
+		q.Add("limit", fmt.Sprintf("%d", limit))
+		q.Add("profile", fmt.Sprintf("%d", profile.ID))
+		q.Add("offset", fmt.Sprintf("%d", offset))
+		req.URL.RawQuery = q.Encode()
+
+		res, err := w.httpClient.Do(req)
+		if err != nil {
+			return transfers, err
+		}
+
+		b, err := io.ReadAll(res.Body)
+		if err != nil {
+			return nil, fmt.Errorf("failed to read response body: %w", err)
+		}
+
+		err = json.Unmarshal(b, &ts)
+		if err != nil {
+			return nil, fmt.Errorf("failed to unmarshal transfers: %w", err)
+		}
+
+		transfers = append(transfers, ts...)
+
+		if len(ts) < limit {
+			break
+		}
+
+		offset += limit
+	}
+
+	return transfers, nil
+}
+
+func newClient(apiKey string) *client {
+	httpClient := &http.Client{
+		Transport: &apiTransport{
+			ApiKey: apiKey,
+		},
+	}
+
+	return &client{
+		httpClient: httpClient,
+	}
+}

pkg/bridge/connectors/wise/config.go

@@ -0,0 +1,13 @@
+package wise
+
+type Config struct {
+	APIKey string `json:"apiKey" yaml:"apiKey" bson:"apiKey"`
+}
+
+func (c Config) Validate() error {
+	if c.APIKey == "" {
+		return ErrMissingAPIKey
+	}
+
+	return nil
+}

pkg/bridge/connectors/wise/errors.go

@@ -0,0 +1,11 @@
+package wise
+
+import "github.com/pkg/errors"
+
+var (
+	// ErrMissingTask is returned when the task is missing.
+	ErrMissingTask = errors.New("task is not implemented")
+
+	// ErrMissingAPIKey is returned when the api key is missing from config.
+	ErrMissingAPIKey = errors.New("missing apiKey from config")
+)

pkg/bridge/connectors/wise/loader.go

@@ -0,0 +1,25 @@
+package wise
+
+import (
+	"github.com/numary/go-libs/sharedlogging"
+	"github.com/numary/payments/pkg/bridge/integration"
+	"github.com/numary/payments/pkg/bridge/task"
+)
+
+// NewLoader creates a new loader.
+func NewLoader() integration.Loader[Config, TaskDefinition] {
+	loader := integration.NewLoaderBuilder[Config, TaskDefinition]("wise").
+		WithLoad(func(logger sharedlogging.Logger, config Config) integration.Connector[TaskDefinition] {
+			return integration.NewConnectorBuilder[TaskDefinition]().
+				WithInstall(func(ctx task.ConnectorContext[TaskDefinition]) error {
+					return ctx.Scheduler().
+						Schedule(
+							TaskDefinition{Name: taskNameFetchProfiles},
+							false)
+				}).
+				WithResolve(resolveTasks(logger, config)).
+				Build()
+		}).Build()
+
+	return loader
+}

pkg/bridge/connectors/wise/task_fetch_profiles.go

@@ -0,0 +1,37 @@
+package wise
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/numary/go-libs/sharedlogging"
+	"github.com/numary/payments/pkg/bridge/task"
+)
+
+func taskFetchProfiles(logger sharedlogging.Logger, client *client) task.Task {
+	return func(
+		ctx context.Context,
+		scheduler task.Scheduler[TaskDefinition],
+	) error {
+		profiles, err := client.getProfiles()
+		if err != nil {
+			return err
+		}
+
+		for _, profile := range profiles {
+			logger.Infof(fmt.Sprintf("scheduling fetch-transfers: %d", profile.ID))
+
+			def := TaskDefinition{
+				Name:      taskNameFetchTransfers,
+				ProfileID: profile.ID,
+			}
+
+			err = scheduler.Schedule(def, false)
+			if err != nil {
+				return err
+			}
+		}
+
+		return nil
+	}
+}

pkg/bridge/connectors/wise/task_fetch_transfers.go

@@ -0,0 +1,65 @@
+package wise
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/numary/go-libs/sharedlogging"
+	payments "github.com/numary/payments/pkg"
+	"github.com/numary/payments/pkg/bridge/ingestion"
+	"github.com/numary/payments/pkg/bridge/task"
+)
+
+func taskFetchTransfers(logger sharedlogging.Logger, client *client, profileID uint64) task.Task {
+	return func(
+		ctx context.Context,
+		scheduler task.Scheduler[TaskDefinition],
+		ingester ingestion.Ingester,
+	) error {
+		transfers, err := client.getTransfers(&profile{
+			ID: profileID,
+		})
+		if err != nil {
+			return err
+		}
+
+		batch := ingestion.Batch{}
+
+		for _, transfer := range transfers {
+			logger.Info(transfer)
+
+			batchElement := ingestion.BatchElement{
+				Referenced: payments.Referenced{
+					Reference: fmt.Sprintf("%d", transfer.ID),
+					Type:      payments.TypeTransfer,
+				},
+				Payment: &payments.Data{
+					Status:        matchTransferStatus(transfer.Status),
+					Scheme:        payments.SchemeOther,
+					InitialAmount: int64(transfer.TargetValue * 100),
+					Asset:         fmt.Sprintf("%s/2", transfer.TargetCurrency),
+					Raw:           transfer,
+				},
+			}
+
+			batch = append(batch, batchElement)
+		}
+
+		return ingester.Ingest(ctx, batch, struct{}{})
+	}
+}
+
+func matchTransferStatus(status string) payments.Status {
+	switch status {
+	case "incoming_payment_waiting", "processing":
+		return payments.StatusPending
+	case "funds_converted", "outgoing_payment_sent":
+		return payments.StatusSucceeded
+	case "bounced_back", "funds_refunded":
+		return payments.StatusFailed
+	case "cancelled":
+		return payments.StatusCancelled
+	}
+
+	return payments.StatusOther
+}

pkg/bridge/connectors/wise/task_resolve.go

@@ -0,0 +1,37 @@
+package wise
+
+import (
+	"fmt"
+
+	"github.com/numary/go-libs/sharedlogging"
+	"github.com/numary/payments/pkg/bridge/task"
+)
+
+const (
+	taskNameFetchTransfers = "fetch-transfers"
+	taskNameFetchProfiles  = "fetch-profiles"
+)
+
+// TaskDefinition is the definition of a task.
+type TaskDefinition struct {
+	Name      string `json:"name" yaml:"name" bson:"name"`
+	ProfileID uint64 `json:"profileID" yaml:"profileID" bson:"profileID"`
+}
+
+func resolveTasks(logger sharedlogging.Logger, config Config) func(taskDefinition TaskDefinition) task.Task {
+	client := newClient(config.APIKey)
+
+	return func(taskDefinition TaskDefinition) task.Task {
+		switch taskDefinition.Name {
+		case taskNameFetchProfiles:
+			return taskFetchProfiles(logger, client)
+		case taskNameFetchTransfers:
+			return taskFetchTransfers(logger, client, taskDefinition.ProfileID)
+		}
+
+		// This should never happen.
+		return func() error {
+			return fmt.Errorf("key '%s': %w", taskDefinition.Name, ErrMissingTask)
+		}
+	}
+}

pkg/bridge/task/task.go

@@ -1,3 +1,3 @@
 package task
 
-type Task interface{}
+type Task any

pkg/payment.go

@@ -35,14 +35,16 @@ const (
 	SchemeAch      Scheme = "ach"
 	SchemeRtp      Scheme = "rtp"
 
-	TypePayIn  = "pay-in"
-	TypePayout = "payout"
-	TypeOther  = "other"
+	TypePayIn    = "pay-in"
+	TypePayout   = "payout"
+	TypeTransfer = "transfer"
+	TypeOther    = "other"
 
 	StatusSucceeded Status = "succeeded"
 	StatusCancelled Status = "cancelled"
 	StatusFailed    Status = "failed"
 	StatusPending   Status = "pending"
+	StatusOther     Status = "other"
 )
 
 type Referenced struct {