feat: add orchestration (#123)

formancehq · May 12, 2023 · 6aa5da8 · 6aa5da8
1 parent 4ea669f
commit 6aa5da8
Show file tree

Hide file tree

Showing 4 changed files with 20 additions and 21 deletions.
diff --git a/pkg/core/script.go b/pkg/core/script.go
@@ -11,14 +11,6 @@ type RunScript struct {
 	Metadata  Metadata `json:"metadata"`
 }
 
-func (s *RunScript) WithDefaultValues() {
-	if s.Timestamp.IsZero() {
-		s.Timestamp = Now()
-	} else {
-		s.Timestamp = s.Timestamp.UTC().Round(DatePrecision)
-	}
-}
-
 type Script struct {
 	Plain string                     `json:"plain"`
 	Vars  map[string]json.RawMessage `json:"vars" swaggertype:"object"`

diff --git a/pkg/ledger/runner/runner.go b/pkg/ledger/runner/runner.go
@@ -56,7 +56,7 @@ func (r *Runner) Execute(
 		return nil, nil, ErrNoScript
 	}
 
-	reserve, err := r.state.Reserve(ctx, state.ReserveRequest{
+	reserve, ts, err := r.state.Reserve(ctx, state.ReserveRequest{
 		Timestamp: script.Timestamp,
 		Reference: script.Reference,
 	})
@@ -65,6 +65,8 @@ func (r *Runner) Execute(
 	}
 	defer reserve.Clear(nil)
 
+	script.Timestamp = *ts
+
 	transaction, logHolder, err := r.execute(ctx, script, logComputer, dryRun)
 	if err != nil {
 		return nil, nil, err

diff --git a/pkg/ledger/state/state.go b/pkg/ledger/state/state.go
@@ -72,17 +72,22 @@ func (s *State) checkConstraints(ctx context.Context, r ReserveRequest) error {
 	return nil
 }
 
-func (s *State) Reserve(ctx context.Context, r ReserveRequest) (*Reserve, error) {
+func (s *State) Reserve(ctx context.Context, r ReserveRequest) (*Reserve, *core.Time, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
 	if err := s.checkConstraints(ctx, r); err != nil {
-		return nil, err
+		return nil, nil, err
+	}
+
+	ts := r.Timestamp
+	if ts.IsZero() {
+		ts = core.Now()
 	}
 
 	ret := &inFlight{
 		reference: r.Reference,
-		timestamp: r.Timestamp,
+		timestamp: ts,
 	}
 	s.inFlights[ret] = struct{}{}
 	if r.Reference != "" {
@@ -101,7 +106,7 @@ func (s *State) Reserve(ctx context.Context, r ReserveRequest) (*Reserve, error)
 	return &Reserve{
 		inFlight: ret,
 		state:    s,
-	}, nil
+	}, &ts, nil
 }
 
 func (s *State) GetMoreRecentTransactionDate() core.Time {

diff --git a/pkg/ledger/state/state_test.go b/pkg/ledger/state/state_test.go
@@ -12,7 +12,7 @@ import (
 
 func TestState(t *testing.T) {
 	state := New(NoOpStore, false, core.Now().Add(-10*time.Second))
-	reserve, err := state.Reserve(context.Background(), ReserveRequest{
+	reserve, _, err := state.Reserve(context.Background(), ReserveRequest{
 		Timestamp: core.Now(),
 	})
 	require.NoError(t, err)
@@ -22,13 +22,13 @@ func TestState(t *testing.T) {
 func TestStateInsertInPastWithNotAllowedPastTimestamp(t *testing.T) {
 	state := New(NoOpStore, false, core.Now().Add(-10*time.Second))
 	now := core.Now()
-	reserve1, err := state.Reserve(context.Background(), ReserveRequest{
+	reserve1, _, err := state.Reserve(context.Background(), ReserveRequest{
 		Timestamp: now,
 	})
 	require.NoError(t, err)
 	defer reserve1.Clear(nil)
 
-	_, err = state.Reserve(context.Background(), ReserveRequest{
+	_, _, err = state.Reserve(context.Background(), ReserveRequest{
 		Timestamp: now.Add(-time.Second),
 	})
 	require.Error(t, err)
@@ -38,13 +38,13 @@ func TestStateInsertInPastWithNotAllowedPastTimestamp(t *testing.T) {
 func TestStateInsertInPastWithAllowPastTimestamps(t *testing.T) {
 	state := New(NoOpStore, true, core.Now().Add(-10*time.Second))
 	now := core.Now()
-	reserve1, err := state.Reserve(context.Background(), ReserveRequest{
+	reserve1, _, err := state.Reserve(context.Background(), ReserveRequest{
 		Timestamp: now,
 	})
 	require.NoError(t, err)
 	defer reserve1.Clear(nil)
 
-	reserve2, err := state.Reserve(context.Background(), ReserveRequest{
+	reserve2, _, err := state.Reserve(context.Background(), ReserveRequest{
 		Timestamp: now.Add(-time.Second),
 	})
 	require.NoError(t, err)
@@ -55,12 +55,12 @@ func TestStateWithError(t *testing.T) {
 	state := New(NoOpStore, false, core.Now().Add(-10*time.Second))
 	now := core.Now()
 
-	_, err := state.Reserve(context.Background(), ReserveRequest{
+	_, _, err := state.Reserve(context.Background(), ReserveRequest{
 		Timestamp: now,
 	})
 	require.NoError(t, err)
 
-	_, err = state.Reserve(context.Background(), ReserveRequest{
+	_, _, err = state.Reserve(context.Background(), ReserveRequest{
 		Timestamp: now.Add(-10 * time.Millisecond),
 	})
 	require.Error(t, err)
@@ -74,7 +74,7 @@ func BenchmarkState(b *testing.B) {
 	eg := errgroup.Group{}
 	for i := 0; i < b.N; i++ {
 		eg.Go(func() error {
-			reserve, err := state.Reserve(context.Background(), ReserveRequest{
+			reserve, _, err := state.Reserve(context.Background(), ReserveRequest{
 				Timestamp: now,
 			})
 			require.NoError(b, err)