/*
Copyright 2021 The Dapr Authors
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
    http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package sidecar

import (
	"fmt"
	"sync/atomic"
	"time"

	"github.com/dapr/dapr/pkg/runtime"
	"github.com/dapr/dapr/pkg/runtime/registry"
	"github.com/dapr/kit/logger"

	"github.com/dapr/components-contrib/tests/certification/embedded"
	"github.com/dapr/components-contrib/tests/certification/flow"

	rtembedded "github.com/dapr/components-contrib/tests/certification/embedded"
	// Go SDK
	dapr "github.com/dapr/go-sdk/client"
)

type (
	Client struct {
		dapr.Client
		registry.ComponentRegistry
		rt     *runtime.DaprRuntime
		errCh  chan error
		closed atomic.Bool
	}

	Sidecar struct {
		appID                    string
		options                  []embedded.Option
		gracefulShutdownDuration time.Duration
	}

	ClientCallback func(client *Client)

	options struct {
		clientCallback ClientCallback
	}

	Option func(o *options)
)

func WithClientCallback(callback ClientCallback) Option {
	return func(o *options) {
		o.clientCallback = callback
	}
}

func GetClient(ctx flow.Context, sidecarName string) *Client {
	var client *Client
	ctx.MustGet(sidecarName, &client)
	return client
}

func Run(appID string, options ...embedded.Option) (string, flow.Runnable, flow.Runnable) {
	return New(appID, options...).ToStep()
}

func New(appID string, options ...embedded.Option) Sidecar {
	return Sidecar{
		appID:   appID,
		options: options,
	}
}

func (s Sidecar) AppID() string {
	return s.appID
}

func (s Sidecar) ToStep() (string, flow.Runnable, flow.Runnable) {
	return s.appID, s.Start, s.Stop
}

func Start(appID string, options ...embedded.Option) flow.Runnable {
	return Sidecar{appID, options, 0}.Start
}

func (s Sidecar) Start(ctx flow.Context) error {
	logContrib := logger.NewLogger("dapr.contrib")

	var options options
	opts := append(rtembedded.CommonComponents(logContrib), s.options...)

	var client Client
	opts = append(opts, func(cfg *runtime.Config) {
		cfg.Registry = cfg.Registry.WithComponentsCallback(func(reg registry.ComponentRegistry) error {
			client.ComponentRegistry = reg
			return nil
		})
	})

	rt, rtConf, err := rtembedded.NewRuntime(ctx, s.appID, opts...)
	if err != nil {
		return err
	}
	s.gracefulShutdownDuration = time.Duration(rtConf.DaprGracefulShutdownSeconds) * time.Second

	client.rt = rt
	client.errCh = make(chan error)

	go func() {
		client.errCh <- rt.Run(ctx)
	}()

	ctx.Cleanup(func() {
		s.Stop(ctx)
	})

	daprClient, err := dapr.NewClientWithPort(fmt.Sprintf("%s", rtConf.DaprAPIGRPCPort))
	if err != nil {
		return err
	}

	client.Client = daprClient

	ctx.Set(s.appID, &client)

	if options.clientCallback != nil {
		options.clientCallback(&client)
	}

	return nil
}

func Stop(appID string) flow.Runnable {
	return Sidecar{appID: appID}.Stop
}

func (s Sidecar) Stop(ctx flow.Context) error {
	var client *Client
	if ctx.Get(s.appID, &client) {
		client.rt.ShutdownWithWait()
		if client.closed.CompareAndSwap(false, true) {
			if err := <-client.errCh; err != nil {
				logger.NewLogger("dapr.contrib").Errorf("error on shutdown: %s", err)
				return err
			}
		}
		return nil
	}

	return nil
}