adapter.go

// Copyright (c) Mainflux
// SPDX-License-Identifier: Apache-2.0

package lora

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"github.com/mainflux/mainflux/pkg/messaging"
)

const protocol = "lora"

var (
	// ErrMalformedMessage indicates malformed LoRa message.
	ErrMalformedMessage = errors.New("malformed message received")

	// ErrNotFoundDev indicates a non-existent route map for a device EUI.
	ErrNotFoundDev = errors.New("route map not found for this device EUI")

	// ErrNotFoundApp indicates a non-existent route map for an application ID.
	ErrNotFoundApp = errors.New("route map not found for this application ID")

	// ErrNotConnected indicates a non-existent route map for a connection.
	ErrNotConnected = errors.New("route map not found for this connection")
)

// Service specifies an API that must be fullfiled by the domain service
// implementation, and all of its decorators (e.g. logging & metrics).
type Service interface {
	// CreateThing creates thingID:devEUI route-map
	CreateThing(ctx context.Context, thingID string, devEUI string) error

	// UpdateThing updates thingID:devEUI route-map
	UpdateThing(ctx context.Context, thingID string, devEUI string) error

	// RemoveThing removes thingID:devEUI route-map
	RemoveThing(ctx context.Context, thingID string) error

	// CreateChannel creates channelID:appID route-map
	CreateChannel(ctx context.Context, chanID string, appID string) error

	// UpdateChannel updates channelID:appID route-map
	UpdateChannel(ctx context.Context, chanID string, appID string) error

	// RemoveChannel removes channelID:appID route-map
	RemoveChannel(ctx context.Context, chanID string) error

	// ConnectThing creates thingID:channelID route-map
	ConnectThing(ctx context.Context, chanID, thingID string) error

	// DisconnectThing removes thingID:channelID route-map
	DisconnectThing(ctx context.Context, chanID, thingID string) error

	// Publish forwards messages from the LoRa MQTT broker to Mainflux Message Broker
	Publish(ctx context.Context, msg *Message) error
}

var _ Service = (*adapterService)(nil)

type adapterService struct {
	publisher  messaging.Publisher
	thingsRM   RouteMapRepository
	channelsRM RouteMapRepository
	connectRM  RouteMapRepository
}

// New instantiates the LoRa adapter implementation.
func New(publisher messaging.Publisher, thingsRM, channelsRM, connectRM RouteMapRepository) Service {
	return &adapterService{
		publisher:  publisher,
		thingsRM:   thingsRM,
		channelsRM: channelsRM,
		connectRM:  connectRM,
	}
}

// Publish forwards messages from Lora MQTT broker to Mainflux Message broker.
func (as *adapterService) Publish(ctx context.Context, m *Message) error {
	// Get route map of lora application
	thingID, err := as.thingsRM.Get(ctx, m.DevEUI)
	if err != nil {
		return ErrNotFoundDev
	}

	// Get route map of lora application
	chanID, err := as.channelsRM.Get(ctx, m.ApplicationID)
	if err != nil {
		return ErrNotFoundApp
	}

	c := fmt.Sprintf("%s:%s", chanID, thingID)
	if _, err := as.connectRM.Get(ctx, c); err != nil {
		return ErrNotConnected
	}

	// Use the SenML message decoded on LoRa Server application if
	// field Object isn't empty. Otherwise, decode standard field Data.
	var payload []byte
	switch m.Object {
	case nil:
		payload, err = base64.StdEncoding.DecodeString(m.Data)
		if err != nil {
			return ErrMalformedMessage
		}
	default:
		jo, err := json.Marshal(m.Object)
		if err != nil {
			return err
		}
		payload = jo
	}

	// Publish on Mainflux Message broker
	msg := messaging.Message{
		Publisher: thingID,
		Protocol:  protocol,
		Channel:   chanID,
		Payload:   payload,
		Created:   time.Now().UnixNano(),
	}

	return as.publisher.Publish(ctx, msg.Channel, &msg)
}

func (as *adapterService) CreateThing(ctx context.Context, thingID string, devEUI string) error {
	return as.thingsRM.Save(ctx, thingID, devEUI)
}

func (as *adapterService) UpdateThing(ctx context.Context, thingID string, devEUI string) error {
	return as.thingsRM.Save(ctx, thingID, devEUI)
}

func (as *adapterService) RemoveThing(ctx context.Context, thingID string) error {
	return as.thingsRM.Remove(ctx, thingID)
}

func (as *adapterService) CreateChannel(ctx context.Context, chanID string, appID string) error {
	return as.channelsRM.Save(ctx, chanID, appID)
}

func (as *adapterService) UpdateChannel(ctx context.Context, chanID string, appID string) error {
	return as.channelsRM.Save(ctx, chanID, appID)
}

func (as *adapterService) RemoveChannel(ctx context.Context, chanID string) error {
	return as.channelsRM.Remove(ctx, chanID)
}

func (as *adapterService) ConnectThing(ctx context.Context, chanID, thingID string) error {
	if _, err := as.channelsRM.Get(ctx, chanID); err != nil {
		return ErrNotFoundApp
	}

	if _, err := as.thingsRM.Get(ctx, thingID); err != nil {
		return ErrNotFoundDev
	}

	c := fmt.Sprintf("%s:%s", chanID, thingID)
	return as.connectRM.Save(ctx, c, c)
}

func (as *adapterService) DisconnectThing(ctx context.Context, chanID, thingID string) error {
	if _, err := as.channelsRM.Get(ctx, chanID); err != nil {
		return ErrNotFoundApp
	}

	if _, err := as.thingsRM.Get(ctx, thingID); err != nil {
		return ErrNotFoundDev
	}

	c := fmt.Sprintf("%s:%s", chanID, thingID)
	return as.connectRM.Remove(ctx, c)
}