client.go

package client

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"fmt"
	"sync"
	"time"

	"github.com/pion/dtls/v2"
	"github.com/plgd-dev/device/client/core"
	"github.com/plgd-dev/device/client/core/otm"
	"github.com/plgd-dev/device/pkg/net/coap"
	"github.com/plgd-dev/go-coap/v2/net/blockwise"
	kitSync "github.com/plgd-dev/kit/v2/sync"
)

type ApplicationCallback = interface {
	GetRootCertificateAuthorities() ([]*x509.Certificate, error)
	GetManufacturerCertificateAuthorities() ([]*x509.Certificate, error)
	GetManufacturerCertificate() (tls.Certificate, error)
}

type subscription = interface {
	Cancel()
	Wait()
}

type Config struct {
	DeviceCacheExpirationSeconds   int64
	ObserverPollingIntervalSeconds uint64 // 0 means 3 seconds

	KeepAliveConnectionTimeoutSeconds uint64 // 0 means keepalive is disabled
	MaxMessageSize                    uint32
	DisablePeerTCPSignalMessageCSMs   bool
	DefaultTransferDurationSeconds    uint64 // 0 means 15 seconds

	// specify one of:
	DeviceOwnershipSDK     *DeviceOwnershipSDKConfig     `yaml:",omitempty"`
	DeviceOwnershipBackend *DeviceOwnershipBackendConfig `yaml:",omitempty"`
}

// NewClientFromConfig constructs a new local client from the proto configuration.
func NewClientFromConfig(cfg *Config, app ApplicationCallback, errors func(error)) (*Client, error) {
	var cacheExpiration time.Duration
	if cfg.DeviceCacheExpirationSeconds > 0 {
		cacheExpiration = time.Second * time.Duration(cfg.DeviceCacheExpirationSeconds)
	}

	observerPollingInterval := time.Second * 3
	if cfg.ObserverPollingIntervalSeconds > 0 {
		observerPollingInterval = time.Second * time.Duration(cfg.ObserverPollingIntervalSeconds)
	}
	dialOpts := make([]coap.DialOptionFunc, 0, 5)
	if cfg.KeepAliveConnectionTimeoutSeconds > 0 {
		dialOpts = append(dialOpts, coap.WithKeepAlive(time.Second*time.Duration(cfg.KeepAliveConnectionTimeoutSeconds)))
	} else {
		dialOpts = append(dialOpts, coap.WithKeepAlive(time.Second*60))
	}
	if cfg.MaxMessageSize > 0 {
		dialOpts = append(dialOpts, coap.WithMaxMessageSize(cfg.MaxMessageSize))
	} else {
		dialOpts = append(dialOpts, coap.WithMaxMessageSize(512*1024))
	}
	if cfg.DisablePeerTCPSignalMessageCSMs {
		dialOpts = append(dialOpts, coap.WithDialDisablePeerTCPSignalMessageCSMs())
	}

	errFn := func(error) {
		// ignore error
	}
	if errors != nil {
		errFn = errors
	}
	dialOpts = append(dialOpts, coap.WithErrors(errFn))
	if cfg.DefaultTransferDurationSeconds > 0 {
		dialOpts = append(dialOpts, coap.WithBlockwise(true, blockwise.SZX1024, time.Second*time.Duration(cfg.DefaultTransferDurationSeconds)))
	} else {
		dialOpts = append(dialOpts, coap.WithBlockwise(true, blockwise.SZX1024, 15*time.Second))
	}

	dialTLS := func(ctx context.Context, addr string, tlsCfg *tls.Config, opts ...coap.DialOptionFunc) (*coap.ClientCloseHandler, error) {
		opts = append(opts, dialOpts...)
		return coap.DialTCPSecure(ctx, addr, tlsCfg, opts...)
	}
	dialDTLS := func(ctx context.Context, addr string, dtlsCfg *dtls.Config, opts ...coap.DialOptionFunc) (*coap.ClientCloseHandler, error) {
		opts = append(opts, dialOpts...)
		return coap.DialUDPSecure(ctx, addr, dtlsCfg, opts...)
	}
	dialTCP := func(ctx context.Context, addr string, opts ...coap.DialOptionFunc) (*coap.ClientCloseHandler, error) {
		opts = append(opts, dialOpts...)
		return coap.DialTCP(ctx, addr, opts...)
	}
	dialUDP := func(ctx context.Context, addr string, opts ...coap.DialOptionFunc) (*coap.ClientCloseHandler, error) {
		opts = append(opts, dialOpts...)
		return coap.DialUDP(ctx, addr, opts...)
	}

	opts := []core.OptionFunc{
		core.WithDialDTLS(dialDTLS),
		core.WithDialTLS(dialTLS),
		core.WithDialTCP(dialTCP),
		core.WithDialUDP(dialUDP),
	}

	deviceOwner, err := NewDeviceOwnerFromConfig(cfg, dialTLS, dialDTLS, app, errors)
	if err != nil {
		return nil, err
	}
	return NewClient(app, deviceOwner, cacheExpiration, observerPollingInterval, errors, opts...)
}

// NewClient constructs a new local client.
func NewClient(
	app ApplicationCallback,
	deviceOwner DeviceOwner,
	cacheExpiration time.Duration,
	observerPollingInterval time.Duration,
	errors func(error),
	opt ...core.OptionFunc,
) (*Client, error) {
	if app == nil {
		return nil, fmt.Errorf("missing application callback")
	}
	if deviceOwner == nil {
		return nil, fmt.Errorf("missing device owner callback")
	}
	tls := core.TLSConfig{
		GetCertificate:            deviceOwner.GetIdentityCertificate,
		GetCertificateAuthorities: deviceOwner.GetIdentityCACerts,
	}
	opt = append(
		[]core.OptionFunc{
			core.WithTLS(&tls),
		},
		opt...,
	)
	if errors != nil {
		opt = append(opt, core.WithErr(errors))
	}
	oc := core.NewClient(opt...)
	pollInterval := time.Second * 10
	if cacheExpiration/2 > pollInterval {
		pollInterval = cacheExpiration / 2
	}
	client := Client{
		client:                  oc,
		app:                     app,
		deviceCache:             NewDeviceCache(cacheExpiration, pollInterval, errors),
		observeResourceCache:    kitSync.NewMap(),
		deviceOwner:             deviceOwner,
		subscriptions:           make(map[string]subscription),
		observerPollingInterval: observerPollingInterval,
		errors:                  errors,
	}
	return &client, nil
}

type ownFunc = func(ctx context.Context, deviceID string, otmClient []otm.Client, discoveryConfiguration core.DiscoveryConfiguration, opts ...core.OwnOption) (string, error)

type DeviceOwner interface {
	Initialization(ctx context.Context) error
	OwnDevice(ctx context.Context, deviceID string, otmTypes []OTMType, discoveryConfiguration core.DiscoveryConfiguration, own ownFunc, opts ...core.OwnOption) (string, error)

	GetIdentityCertificate() (tls.Certificate, error)
	GetIdentityCACerts() ([]*x509.Certificate, error)
}

// Client uses the underlying OCF local client.
type Client struct {
	app    ApplicationCallback
	client *core.Client

	deviceCache *DeviceCache

	observeResourceCache    *kitSync.Map
	observerPollingInterval time.Duration

	deviceOwner DeviceOwner

	subscriptionsLock sync.Mutex
	subscriptions     map[string]subscription

	disableUDPEndpoints bool
	errors              func(error)
}

func (c *Client) popSubscriptions() map[string]subscription {
	c.subscriptionsLock.Lock()
	defer c.subscriptionsLock.Unlock()
	s := c.subscriptions
	c.subscriptions = make(map[string]subscription)
	return s
}

func (c *Client) popSubscription(id string) (subscription, error) {
	c.subscriptionsLock.Lock()
	defer c.subscriptionsLock.Unlock()
	v, ok := c.subscriptions[id]
	if !ok {
		return nil, fmt.Errorf("cannot find observation %v", id)
	}
	delete(c.subscriptions, id)
	return v, nil
}

func (c *Client) insertSubscription(id string, s subscription) {
	c.subscriptionsLock.Lock()
	defer c.subscriptionsLock.Unlock()
	c.subscriptions[id] = s
}

func (c *Client) CoreClient() *core.Client {
	return c.client
}

// Close clears all connections and spawned goroutines by client.
func (c *Client) Close(ctx context.Context) error {
	for _, s := range c.popSubscriptions() {
		s.Cancel()
	}
	return c.deviceCache.Close(ctx)
}

func NewDeviceOwnerFromConfig(cfg *Config, dialTLS core.DialTLS, dialDTLS core.DialDTLS, app ApplicationCallback, errors func(error)) (DeviceOwner, error) {
	if cfg.DeviceOwnershipSDK != nil {
		c, err := NewDeviceOwnershipSDKFromConfig(app, dialTLS, dialDTLS, cfg.DeviceOwnershipSDK)
		if err != nil {
			return nil, fmt.Errorf("cannot create sdk signers: %w", err)
		}
		return c, nil
	}
	if cfg.DeviceOwnershipBackend != nil {
		c, err := NewDeviceOwnershipBackendFromConfig(app, dialTLS, dialDTLS, cfg.DeviceOwnershipBackend, errors)
		if err != nil {
			return nil, fmt.Errorf("cannot create server signers: %w", err)
		}
		return c, nil
	}
	return NewDeviceOwnershipNone(), nil
}