feat: mqtt + nats communication with device status and info

backend/services/controller/internal/db/device.go

@@ -38,7 +38,6 @@ type Device struct {
 	Websockets   Status
 }
 
-// TODO: don't change device status of other MTP
 func (d *Database) CreateDevice(device Device) error {
 	var result bson.M
 	var deviceExistent Device

backend/services/mtp/adapter/.env.local

@@ -0,0 +1 @@
+MONGO_URI="mongodb://172.16.235.2:27017"

backend/services/mtp/adapter/README.md

@@ -0,0 +1,2 @@
+- Abstracts all other mtps existence
+- Saves devices info and status

backend/services/mtp/adapter/cmd/adapter/main.go

@@ -0,0 +1,33 @@
+package main
+
+import (
+	"log"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"github.com/OktopUSP/oktopus/backend/services/mtp/adapter/internal/config"
+	"github.com/OktopUSP/oktopus/backend/services/mtp/adapter/internal/db"
+	"github.com/OktopUSP/oktopus/backend/services/mtp/adapter/internal/events"
+	"github.com/OktopUSP/oktopus/backend/services/mtp/adapter/internal/events/handler"
+	"github.com/OktopUSP/oktopus/backend/services/mtp/adapter/internal/nats"
+)
+
+func main() {
+	done := make(chan os.Signal, 1)
+	signal.Notify(done, syscall.SIGINT, syscall.SIGTERM)
+
+	c := config.NewConfig()
+
+	js, nc := nats.StartNatsClient(c.Nats)
+
+	db := db.NewDatabase(c.Mongo.Ctx, c.Mongo.Uri)
+
+	handler := handler.NewHandler(nc, js, db, c.ControllerId)
+
+	events.StartEventsListener(c.Nats.Ctx, js, handler)
+
+	<-done
+
+	log.Println("mtp adapter is shutting down...")
+}

backend/services/mtp/adapter/go.mod

@@ -0,0 +1,27 @@
+module github.com/OktopUSP/oktopus/backend/services/mtp/adapter
+
+go 1.22.0
+
+require (
+	github.com/google/uuid v1.6.0
+	github.com/joho/godotenv v1.5.1
+	github.com/nats-io/nats.go v1.33.1
+	go.mongodb.org/mongo-driver v1.14.0
+	google.golang.org/protobuf v1.32.0
+)
+
+require (
+	github.com/golang/snappy v0.0.1 // indirect
+	github.com/klauspost/compress v1.17.2 // indirect
+	github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe // indirect
+	github.com/nats-io/nkeys v0.4.7 // indirect
+	github.com/nats-io/nuid v1.0.1 // indirect
+	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
+	github.com/xdg-go/scram v1.1.2 // indirect
+	github.com/xdg-go/stringprep v1.0.4 // indirect
+	github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d // indirect
+	golang.org/x/crypto v0.18.0 // indirect
+	golang.org/x/sync v0.1.0 // indirect
+	golang.org/x/sys v0.16.0 // indirect
+	golang.org/x/text v0.14.0 // indirect
+)

backend/services/mtp/adapter/go.sum

@@ -0,0 +1,66 @@
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
+github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
+github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=
+github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4=
+github.com/klauspost/compress v1.17.2 h1:RlWWUY/Dr4fL8qk9YG7DTZ7PDgME2V4csBXA8L/ixi4=
+github.com/klauspost/compress v1.17.2/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
+github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe h1:iruDEfMl2E6fbMZ9s0scYfZQ84/6SPL6zC8ACM2oIL0=
+github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe/go.mod h1:wL8QJuTMNUDYhXwkmfOly8iTdp5TEcJFWZD2D7SIkUc=
+github.com/nats-io/nats.go v1.33.1 h1:8TxLZZ/seeEfR97qV0/Bl939tpDnt2Z2fK3HkPypj70=
+github.com/nats-io/nats.go v1.33.1/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8=
+github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI=
+github.com/nats-io/nkeys v0.4.7/go.mod h1:kqXRgRDPlGy7nGaEDMuYzmiJCIAAWDK0IMBtDmGD0nc=
+github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
+github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
+github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c=
+github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
+github.com/xdg-go/scram v1.1.2 h1:FHX5I5B4i4hKRVRBCFRxq1iQRej7WO3hhBuJf+UUySY=
+github.com/xdg-go/scram v1.1.2/go.mod h1:RT/sEzTbU5y00aCK8UOx6R7YryM0iF1N2MOmC3kKLN4=
+github.com/xdg-go/stringprep v1.0.4 h1:XLI/Ng3O1Atzq0oBs3TWm+5ZVgkq2aqdlvP9JtoZ6c8=
+github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM=
+github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d h1:splanxYIlg+5LfHAM6xpdFEAYOk8iySO56hMFq6uLyA=
+github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d/go.mod h1:rHwXgn7JulP+udvsHwJoVG1YGAP6VLg4y9I5dyZdqmA=
+github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
+go.mongodb.org/mongo-driver v1.14.0 h1:P98w8egYRjYe3XDjxhYJagTokP/H6HzlsnojRgZRd80=
+go.mongodb.org/mongo-driver v1.14.0/go.mod h1:Vzb0Mk/pa7e6cWw85R4F/endUC3u0U9jGcNU603k65c=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
+golang.org/x/crypto v0.18.0 h1:PGVlW0xEltQnzFZ55hkuX5+KLyrMYhHld1YHO4AKcdc=
+golang.org/x/crypto v0.18.0/go.mod h1:R0j02AL6hcrfOiy9T4ZYp/rcWeMxM3L6QYxlOuEG1mg=
+golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o=
+golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.16.0 h1:xWw16ngr6ZMtmxDyKyIgsE93KNKz5HKmMa3b8ALHidU=
+golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
+golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
+golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
+golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
+golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+google.golang.org/protobuf v1.32.0 h1:pPC6BG5ex8PDFnkbrGU3EixyhKcQ2aDuBS36lqK/C7I=
+google.golang.org/protobuf v1.32.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=

backend/services/mtp/adapter/internal/config/config.go

@@ -0,0 +1,107 @@
+package config
+
+import (
+	"context"
+	"flag"
+	"log"
+	"os"
+	"strconv"
+
+	"github.com/joho/godotenv"
+)
+
+const LOCAL_ENV = ".env.local"
+
+type Nats struct {
+	Url                string
+	Name               string
+	VerifyCertificates bool
+	Ctx                context.Context
+}
+
+type Mongo struct {
+	Uri string
+	Ctx context.Context
+}
+
+type Config struct {
+	Nats         Nats
+	Mongo        Mongo
+	ControllerId string
+}
+
+func NewConfig() *Config {
+
+	loadEnvVariables()
+	log.SetFlags(log.LstdFlags | log.Lshortfile)
+
+	natsUrl := flag.String("nats_url", lookupEnvOrString("NATS_URL", "nats://localhost:4222"), "url for nats server")
+	natsName := flag.String("nats_name", lookupEnvOrString("NATS_NAME", "adapter"), "name for nats client")
+	natsVerifyCertificates := flag.Bool("nats_verify_certificates", lookupEnvOrBool("NATS_VERIFY_CERTIFICATES", false), "verify validity of certificates from nats server")
+	mongoUri := flag.String("mongo_uri", lookupEnvOrString("MONGO_URI", "mongodb://localhost:27017"), "uri for mongodb server")
+	controllerId := flag.String("controller_id", lookupEnvOrString("CONTROLLER_ID", "oktopusController"), "usp controller endpoint id")
+	flHelp := flag.Bool("help", false, "Help")
+
+	/*
+		App variables priority:
+		1º - Flag through command line.
+		2º - Env variables.
+		3º - Default flag value.
+	*/
+
+	flag.Parse()
+
+	if *flHelp {
+		flag.Usage()
+		os.Exit(0)
+	}
+
+	ctx := context.TODO()
+
+	return &Config{
+		Nats: Nats{
+			Url:                *natsUrl,
+			Name:               *natsName,
+			VerifyCertificates: *natsVerifyCertificates,
+			Ctx:                ctx,
+		},
+		Mongo: Mongo{
+			Uri: *mongoUri,
+			Ctx: ctx,
+		},
+		ControllerId: *controllerId,
+	}
+}
+
+func loadEnvVariables() {
+	err := godotenv.Load()
+
+	if _, err := os.Stat(LOCAL_ENV); err == nil {
+		_ = godotenv.Overload(LOCAL_ENV)
+		log.Printf("Loaded variables from '%s'", LOCAL_ENV)
+	}
+
+	if err != nil {
+		log.Println("Error to load environment variables:", err)
+	} else {
+		log.Println("Loaded variables from '.env'")
+	}
+}
+
+func lookupEnvOrString(key string, defaultVal string) string {
+	if val, _ := os.LookupEnv(key); val != "" {
+		return val
+	}
+	return defaultVal
+}
+
+func lookupEnvOrBool(key string, defaultVal bool) bool {
+	if val, _ := os.LookupEnv(key); val != "" {
+		v, err := strconv.ParseBool(val)
+		if err != nil {
+			log.Fatalf("LookupEnvOrBool[%s]: %v", key, err)
+		}
+		return v
+	}
+	return defaultVal
+}

backend/services/mtp/adapter/internal/db/db.go

@@ -0,0 +1,57 @@
+package db
+
+import (
+	"context"
+	"log"
+	"sync"
+
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+)
+
+type Database struct {
+	client  *mongo.Client
+	devices *mongo.Collection
+	ctx     context.Context
+	m       *sync.Mutex
+}
+
+func NewDatabase(ctx context.Context, mongoUri string) Database {
+	var db Database
+
+	clientOptions := options.Client().ApplyURI(mongoUri)
+	client, err := mongo.Connect(ctx, clientOptions)
+	if err != nil {
+		log.Fatal(err)
+	}
+	db.client = client
+
+	log.Println("Trying to ping Mongo database...")
+	err = client.Ping(ctx, nil)
+	if err != nil {
+		log.Fatal("Couldn't connect to MongoDB --> ", err)
+	}
+
+	log.Println("Connected to MongoDB-->", mongoUri)
+
+	devices := client.Database("adapter").Collection("devices")
+	createIndexes(ctx, devices)
+
+	db.devices = devices
+	db.ctx = ctx
+	db.m = &sync.Mutex{}
+
+	return db
+}
+
+func createIndexes(ctx context.Context, devices *mongo.Collection) {
+	indexField := bson.M{"sn": 1}
+	_, err := devices.Indexes().CreateOne(ctx, mongo.IndexModel{
+		Keys:    indexField,
+		Options: options.Index().SetUnique(true),
+	})
+	if err != nil {
+		log.Println("ERROR to create index in database:", err)
+	}
+}

backend/services/mtp/adapter/internal/db/device.go

@@ -0,0 +1,148 @@
+package db
+
+import (
+	"log"
+
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+)
+
+type MTP int32
+
+const (
+	UNDEFINED MTP = iota
+	MQTT
+	STOMP
+	WEBSOCKETS
+)
+
+type Status uint8
+
+const (
+	Offline Status = iota
+	Associating
+	Online
+)
+
+type Device struct {
+	SN           string
+	Model        string
+	Customer     string
+	Vendor       string
+	Version      string
+	ProductClass string
+	Status       Status
+	Mqtt         Status
+	Stomp        Status
+	Websockets   Status
+}
+
+func (d *Database) CreateDevice(device Device) error {
+	var result bson.M
+	var deviceExistent Device
+
+	d.m.Lock()
+	defer d.m.Unlock()
+
+	/* ------------------ Do not overwrite status of other mtp ------------------ */
+	err := d.devices.FindOne(d.ctx, bson.D{{"sn", device.SN}}, nil).Decode(&deviceExistent)
+	if err == nil {
+		if deviceExistent.Mqtt == Online {
+			device.Mqtt = Online
+		}
+		if deviceExistent.Stomp == Online {
+			device.Stomp = Online
+		}
+		if deviceExistent.Websockets == Online {
+			device.Websockets = Online
+		}
+	} else {
+		if err != nil && err != mongo.ErrNoDocuments {
+			log.Println(err)
+			return err
+		}
+	}
+	/* -------------------------------------------------------------------------- */
+
+	callback := func(sessCtx mongo.SessionContext) (interface{}, error) {
+		// Important: You must pass sessCtx as the Context parameter to the operations for them to be executed in the
+		// transaction.
+		opts := options.FindOneAndReplace().SetUpsert(true)
+
+		err = d.devices.FindOneAndReplace(d.ctx, bson.D{{"sn", device.SN}}, device, opts).Decode(&result)
+		if err != nil {
+			if err == mongo.ErrNoDocuments {
+				log.Printf("New device %s added to database", device.SN)
+				return nil, nil
+			}
+			return nil, err
+		}
+		log.Printf("Device %s already existed, and got replaced for new info", device.SN)
+		return nil, nil
+	}
+
+	session, err := d.client.StartSession()
+	if err != nil {
+		return err
+	}
+	defer session.EndSession(d.ctx)
+
+	_, err = session.WithTransaction(d.ctx, callback)
+	if err != nil {
+		return err
+	}
+	return err
+}
+func (d *Database) RetrieveDevices(filter bson.A) ([]Device, error) {
+	cursor, err := d.devices.Aggregate(d.ctx, filter)
+
+	var results []Device
+
+	for cursor.Next(d.ctx) {
+		var device Device
+
+		err := cursor.Decode(&device)
+		if err != nil {
+			log.Println("Error to decode device info fields")
+			continue
+		}
+
+		results = append(results, device)
+	}
+
+	return results, err
+}
+
+func (d *Database) RetrieveDevice(sn string) (Device, error) {
+	var result Device
+	//TODO: filter devices by user ownership
+	err := d.devices.FindOne(d.ctx, bson.D{{"sn", sn}}, nil).Decode(&result)
+	if err != nil {
+		log.Println(err)
+	}
+	return result, err
+}
+
+func (d *Database) RetrieveDevicesCount(filter bson.M) (int64, error) {
+	count, err := d.devices.CountDocuments(d.ctx, filter)
+	return count, err
+}
+
+func (d *Database) DeleteDevice() {
+
+}
+
+func (m MTP) String() string {
+	switch m {
+	case UNDEFINED:
+		return "unknown"
+	case MQTT:
+		return "mqtt"
+	case STOMP:
+		return "stomp"
+	case WEBSOCKETS:
+		return "websockets"
+	}
+	return "unknown"
+}

backend/services/mtp/adapter/internal/db/info.go

@@ -0,0 +1,97 @@
+package db
+
+import (
+	"log"
+
+	"go.mongodb.org/mongo-driver/bson"
+)
+
+type VendorsCount struct {
+	Vendor string `bson:"_id" json:"vendor"`
+	Count  int    `bson:"count" json:"count"`
+}
+
+type ProductClassCount struct {
+	ProductClass string `bson:"_id" json:"productClass"`
+	Count        int    `bson:"count" json:"count"`
+}
+
+type StatusCount struct {
+	Status int `bson:"_id" json:"status"`
+	Count  int `bson:"count" json:"count"`
+}
+
+func (d *Database) RetrieveVendorsInfo() ([]VendorsCount, error) {
+	var results []VendorsCount
+	cursor, err := d.devices.Aggregate(d.ctx, []bson.M{
+		{
+			"$group": bson.M{
+				"_id":   "$vendor",
+				"count": bson.M{"$sum": 1},
+			},
+		},
+	})
+	if err != nil {
+		log.Println(err)
+		return nil, err
+	}
+	defer cursor.Close(d.ctx)
+	if err := cursor.All(d.ctx, &results); err != nil {
+		log.Println(err)
+		return nil, err
+	}
+	// for _, result := range results {
+	// 	log.Println(result)
+	// }
+	return results, nil
+}
+
+func (d *Database) RetrieveStatusInfo() ([]StatusCount, error) {
+	var results []StatusCount
+	cursor, err := d.devices.Aggregate(d.ctx, []bson.M{
+		{
+			"$group": bson.M{
+				"_id":   "$status",
+				"count": bson.M{"$sum": 1},
+			},
+		},
+	})
+	if err != nil {
+		log.Println(err)
+		return nil, err
+	}
+	defer cursor.Close(d.ctx)
+	if err := cursor.All(d.ctx, &results); err != nil {
+		log.Println(err)
+		return nil, err
+	}
+	// for _, result := range results {
+	// 	log.Println(result)
+	// }
+	return results, nil
+}
+
+func (d *Database) RetrieveProductsClassInfo() ([]ProductClassCount, error) {
+	var results []ProductClassCount
+	cursor, err := d.devices.Aggregate(d.ctx, []bson.M{
+		{
+			"$group": bson.M{
+				"_id":   "$productclass",
+				"count": bson.M{"$sum": 1},
+			},
+		},
+	})
+	if err != nil {
+		log.Println(err)
+		return nil, err
+	}
+	defer cursor.Close(d.ctx)
+	if err := cursor.All(d.ctx, &results); err != nil {
+		log.Println(err)
+		return nil, err
+	}
+	// for _, result := range results {
+	// 	log.Println(result)
+	// }
+	return results, nil
+}

backend/services/mtp/adapter/internal/db/status.go

@@ -0,0 +1,69 @@
+package db
+
+import (
+	"log"
+
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/bson/primitive"
+	"go.mongodb.org/mongo-driver/mongo"
+)
+
+func (d *Database) UpdateStatus(sn string, status Status, mtp MTP) error {
+	var result Device
+
+	d.m.Lock()
+	defer d.m.Unlock()
+	err := d.devices.FindOne(d.ctx, bson.D{{"sn", sn}}, nil).Decode(&result)
+	if err != nil {
+		log.Println(err)
+	}
+
+	//TODO: abolish this logic, find another approach, microservices design maybe?
+	/*
+		In case the device status is online, we must check if the mtp
+		changing is going to affect the global status. In case it does,
+		we must update the global status accordingly.
+	*/
+
+	/*
+		mix the existent device status to the updated one
+	*/
+	switch mtp {
+	case MQTT:
+		result.Mqtt = status
+	case STOMP:
+		result.Stomp = status
+	case WEBSOCKETS:
+		result.Websockets = status
+	}
+
+	/*
+		check if the global status needs update
+	*/
+	var globalStatus primitive.E
+	if result.Mqtt == Offline && result.Stomp == Offline && result.Websockets == Offline {
+		globalStatus = primitive.E{"status", Offline}
+	}
+	if result.Mqtt == Online || result.Stomp == Online || result.Websockets == Online {
+		globalStatus = primitive.E{"status", Online}
+	}
+
+	_, err = d.devices.UpdateOne(d.ctx, bson.D{{"sn", sn}}, bson.D{
+		{
+			"$set", bson.D{
+				{mtp.String(), status},
+				globalStatus,
+			},
+		},
+	})
+
+	if err != nil {
+		if err == mongo.ErrNoDocuments {
+			log.Printf("Device %s is not mapped into database", sn)
+			return nil
+		}
+		log.Println(err)
+	}
+	log.Printf("%s is now offline.", sn)
+	return err
+}

backend/services/mtp/adapter/internal/events/events.go

@@ -0,0 +1,61 @@
+package events
+
+import (
+	"context"
+	"log"
+	"strings"
+
+	"github.com/OktopUSP/oktopus/backend/services/mtp/adapter/internal/events/handler"
+	"github.com/OktopUSP/oktopus/backend/services/mtp/adapter/internal/nats"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+func StartEventsListener(ctx context.Context, js jetstream.JetStream, h handler.Handler) {
+
+	events := []string{
+		nats.MQTT_STREAM_NAME,
+		nats.WS_STREAM_NAME,
+		nats.STOMP_STREAM_NAME,
+		nats.LORA_STREAM_NAME,
+		nats.OPC_STREAM_NAME,
+	}
+
+	for _, event := range events {
+		go func() {
+			consumer, err := js.Consumer(ctx, event, event)
+			if err != nil {
+				log.Fatalf("Failed to get consumer: %v", err)
+			}
+			messages, err := consumer.Messages()
+			if err != nil {
+				log.Fatalf("Failed to get consumer messages: %v", err)
+			}
+			defer messages.Stop()
+			for {
+				msg, err := messages.Next()
+				if err != nil {
+					log.Println("Error to get next message:", err)
+					continue
+				}
+
+				data := msg.Data()
+
+				log.Printf("Received message, subject: %s", msg.Subject())
+
+				subject := strings.Split(msg.Subject(), ".")
+				msgType := subject[len(subject)-1]
+				device := subject[len(subject)-2]
+
+				switch msgType {
+				case "status":
+					h.HandleDeviceStatus(device, msg.Subject(), data)
+				case "info":
+					h.HandleDeviceInfo(device, msg.Subject(), data, event)
+				default:
+					//ignoreMsg(msg.Subject(), "status", msg.Data())
+				}
+				msg.Ack()
+			}
+		}()
+	}
+}

backend/services/mtp/adapter/internal/events/handler/handler.go

@@ -0,0 +1,30 @@
+package handler
+
+import (
+	"github.com/OktopUSP/oktopus/backend/services/mtp/adapter/internal/db"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+const (
+	ONLINE = iota
+	OFFLINE
+)
+
+const NATS_SUBJ_PREFIX = "mqtt-adapter.usp.v1."
+
+type Handler struct {
+	nc  *nats.Conn
+	js  jetstream.JetStream
+	db  db.Database
+	cid string
+}
+
+func NewHandler(nc *nats.Conn, js jetstream.JetStream, d db.Database, cid string) Handler {
+	return Handler{
+		nc:  nc,
+		js:  js,
+		db:  d,
+		cid: cid,
+	}
+}

backend/services/mtp/adapter/internal/events/handler/info.go

@@ -0,0 +1,68 @@
+package handler
+
+import (
+	"log"
+
+	"github.com/OktopUSP/oktopus/backend/services/mtp/adapter/internal/db"
+	"github.com/OktopUSP/oktopus/backend/services/mtp/adapter/internal/nats"
+	"github.com/OktopUSP/oktopus/backend/services/mtp/adapter/internal/usp/usp_msg"
+	"github.com/OktopUSP/oktopus/backend/services/mtp/adapter/internal/usp/usp_record"
+	"google.golang.org/protobuf/proto"
+)
+
+func (h *Handler) HandleDeviceInfo(device, subject string, data []byte, mtp string) {
+	log.Printf("Device %s info", device)
+	deviceInfo := parseDeviceInfoMsg(device, subject, data, getMtp(mtp))
+	err := h.db.CreateDevice(deviceInfo)
+	if err != nil {
+		log.Printf("Failed to create device: %v", err)
+	}
+}
+
+func getMtp(mtp string) db.MTP {
+	switch mtp {
+	case nats.MQTT_STREAM_NAME:
+		return db.MQTT
+	case nats.WS_STREAM_NAME:
+		return db.WEBSOCKETS
+	case nats.STOMP_STREAM_NAME:
+		return db.STOMP
+	default:
+		return db.MTP(0)
+	}
+}
+
+func parseDeviceInfoMsg(sn, subject string, data []byte, mtp db.MTP) db.Device {
+	var record usp_record.Record
+	var message usp_msg.Msg
+
+	err := proto.Unmarshal(data, &record)
+	if err != nil {
+		log.Fatal(err)
+	}
+	err = proto.Unmarshal(record.GetNoSessionContext().Payload, &message)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	var device db.Device
+	msg := message.Body.MsgBody.(*usp_msg.Body_Response).Response.GetGetResp()
+
+	device.Vendor = msg.ReqPathResults[0].ResolvedPathResults[0].ResultParams["Manufacturer"]
+	device.Model = msg.ReqPathResults[1].ResolvedPathResults[0].ResultParams["ModelName"]
+	device.Version = msg.ReqPathResults[2].ResolvedPathResults[0].ResultParams["SoftwareVersion"]
+	device.ProductClass = msg.ReqPathResults[4].ResolvedPathResults[0].ResultParams["ProductClass"]
+	device.SN = sn
+	switch db.MTP(mtp) {
+	case db.MQTT:
+		device.Mqtt = db.Online
+	case db.WEBSOCKETS:
+		device.Websockets = db.Online
+	case db.STOMP:
+		device.Stomp = db.Online
+	}
+
+	device.Status = db.Online
+
+	return device
+}

backend/services/mtp/adapter/internal/events/handler/status.go

@@ -0,0 +1,69 @@
+package handler
+
+import (
+	"log"
+	"strconv"
+
+	"github.com/OktopUSP/oktopus/backend/services/mtp/adapter/internal/db"
+	"github.com/OktopUSP/oktopus/backend/services/mtp/adapter/internal/usp"
+	"github.com/OktopUSP/oktopus/backend/services/mtp/adapter/internal/usp/usp_msg"
+	"google.golang.org/protobuf/proto"
+)
+
+func (h *Handler) HandleDeviceStatus(device, subject string, data []byte) {
+	payload, err := strconv.Atoi(string(data))
+	if err != nil {
+		log.Printf("Status subject payload message error %q", err)
+	}
+
+	switch payload {
+	case ONLINE:
+		h.deviceOnline(device)
+	case OFFLINE:
+		h.deviceOffline(device)
+	default:
+		ignoreMsg(subject, "status", data)
+	}
+}
+
+func (h *Handler) deviceOnline(device string) {
+
+	log.Printf("Device %s is online", device)
+
+	msg := usp.NewGetMsg(usp_msg.Get{
+		ParamPaths: []string{
+			"Device.DeviceInfo.Manufacturer",
+			"Device.DeviceInfo.ModelName",
+			"Device.DeviceInfo.SoftwareVersion",
+			"Device.DeviceInfo.SerialNumber",
+			"Device.DeviceInfo.ProductClass",
+		},
+		MaxDepth: 1,
+	})
+
+	payload, _ := proto.Marshal(&msg)
+	record := usp.NewUspRecord(payload, device, h.cid)
+
+	tr369Message, err := proto.Marshal(&record)
+	if err != nil {
+		log.Fatalln("Failed to encode tr369 record:", err)
+	}
+
+	err = h.nc.Publish(NATS_SUBJ_PREFIX+device+".info", tr369Message)
+	if err != nil {
+		log.Printf("Failed to publish online device message: %v", err)
+	}
+}
+
+func (h *Handler) deviceOffline(device string) {
+	log.Printf("Device %s is offline", device)
+
+	err := h.db.UpdateStatus(device, db.Offline, db.MQTT)
+	if err != nil {
+		log.Fatal(err)
+	}
+}
+
+func ignoreMsg(subject, ctx string, data []byte) {
+	log.Printf("Unknown message of %s received, subject: %s, payload: %s. Ignored...", ctx, subject, string(data))
+}

backend/services/mtp/adapter/internal/nats/nats.go

@@ -0,0 +1,132 @@
+package nats
+
+import (
+	"context"
+	"errors"
+	"log"
+	"time"
+
+	"github.com/OktopUSP/oktopus/backend/services/mtp/adapter/internal/config"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+const (
+	MQTT_ADAPTER_STREAM_NAME = "mqtt-adapter"
+	MQTT_STREAM_NAME         = "mqtt"
+	WS_STREAM_NAME           = "ws"
+	STOMP_STREAM_NAME        = "stomp"
+	LORA_STREAM_NAME         = "lora"
+	OPC_STREAM_NAME          = "opc"
+)
+
+func StartNatsClient(c config.Nats) (jetstream.JetStream, *nats.Conn) {
+
+	var (
+		nc  *nats.Conn
+		err error
+	)
+
+	opts := defineOptions(c)
+
+	for {
+		nc, err = nats.Connect(c.Url, opts...)
+		if err != nil {
+			time.Sleep(5 * time.Second)
+			continue
+		}
+		break
+	}
+	log.Printf("Successfully connected to NATS server %s", c.Url)
+
+	js, err := jetstream.New(nc)
+	if err != nil {
+		log.Fatalf("Failed to create JetStream client: %v", err)
+	}
+
+	streams := defineStreams()
+	err = createStreams(c.Ctx, js, streams)
+	if err != nil {
+		log.Fatalf("Failed to create Consumer: %v", err)
+	}
+
+	consumers := defineConsumers()
+	err = createConsumers(c.Ctx, js, consumers)
+	if err != nil {
+		log.Fatalf("Failed to create Consumer: %v", err)
+	}
+
+	return js, nc
+}
+
+func createStreams(ctx context.Context, js jetstream.JetStream, streams []string) error {
+	for _, stream := range streams {
+		_, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
+			Name:        stream,
+			Description: "Stream for " + stream + " messages",
+			Subjects:    []string{stream + ".>"},
+			Retention:   jetstream.InterestPolicy,
+		})
+		if err != nil {
+			return errors.New(err.Error() + " | consumer:" + stream)
+		}
+	}
+	return nil
+}
+
+func createConsumers(ctx context.Context, js jetstream.JetStream, consumers []string) error {
+	for _, consumer := range consumers {
+		_, err := js.CreateOrUpdateConsumer(ctx, consumer, jetstream.ConsumerConfig{
+			Name:        consumer,
+			Description: "Consumer for " + consumer + " messages",
+			AckPolicy:   jetstream.AckExplicitPolicy,
+			Durable:     consumer,
+		})
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func defineStreams() []string {
+	return []string{
+		MQTT_STREAM_NAME,
+		WS_STREAM_NAME,
+		STOMP_STREAM_NAME,
+		LORA_STREAM_NAME,
+		OPC_STREAM_NAME,
+	}
+}
+
+func defineConsumers() []string {
+	return []string{
+		MQTT_STREAM_NAME,
+		WS_STREAM_NAME,
+		STOMP_STREAM_NAME,
+		LORA_STREAM_NAME,
+		OPC_STREAM_NAME,
+	}
+}
+
+func defineOptions(c config.Nats) []nats.Option {
+	var opts []nats.Option
+
+	opts = append(opts, nats.Name(c.Name))
+	opts = append(opts, nats.MaxReconnects(-1))
+	opts = append(opts, nats.ReconnectWait(5*time.Second))
+	opts = append(opts, nats.DisconnectErrHandler(func(nc *nats.Conn, err error) {
+		log.Printf("Got disconnected! Reason: %q\n", err)
+	}))
+	opts = append(opts, nats.ReconnectHandler(func(nc *nats.Conn) {
+		log.Printf("Got reconnected to %v!\n", nc.ConnectedUrl())
+	}))
+	opts = append(opts, nats.ClosedHandler(func(nc *nats.Conn) {
+		log.Printf("Connection closed. Reason: %q\n", nc.LastError())
+	}))
+	if c.VerifyCertificates {
+		opts = append(opts, nats.RootCAs())
+	}
+
+	return opts
+}

backend/services/mtp/adapter/internal/usp/usp.go

@@ -0,0 +1,149 @@
+package usp
+
+import (
+	"github.com/OktopUSP/oktopus/backend/services/mtp/adapter/internal/usp/usp_msg"
+	"github.com/OktopUSP/oktopus/backend/services/mtp/adapter/internal/usp/usp_record"
+	"github.com/google/uuid"
+)
+
+const VERSION = "1.0"
+
+func NewUspRecord(p []byte, toId, fromId string) usp_record.Record {
+	return usp_record.Record{
+		Version:         VERSION,
+		ToId:            toId,
+		FromId:          fromId,
+		PayloadSecurity: usp_record.Record_PLAINTEXT,
+		RecordType: &usp_record.Record_NoSessionContext{
+			NoSessionContext: &usp_record.NoSessionContextRecord{
+				Payload: p,
+			},
+		},
+	}
+}
+
+func NewCreateMsg(createStuff usp_msg.Add) usp_msg.Msg {
+	return usp_msg.Msg{
+		Header: &usp_msg.Header{
+			MsgId:   uuid.NewString(),
+			MsgType: usp_msg.Header_ADD,
+		},
+		Body: &usp_msg.Body{
+			MsgBody: &usp_msg.Body_Request{
+				Request: &usp_msg.Request{
+					ReqType: &usp_msg.Request_Add{
+						Add: &createStuff,
+					},
+				},
+			},
+		},
+	}
+}
+
+func NewGetMsg(getStuff usp_msg.Get) usp_msg.Msg {
+	return usp_msg.Msg{
+		Header: &usp_msg.Header{
+			MsgId:   uuid.NewString(),
+			MsgType: usp_msg.Header_GET,
+		},
+		Body: &usp_msg.Body{
+			MsgBody: &usp_msg.Body_Request{
+				Request: &usp_msg.Request{
+					ReqType: &usp_msg.Request_Get{
+						Get: &getStuff,
+					},
+				},
+			},
+		},
+	}
+}
+
+func NewDelMsg(getStuff usp_msg.Delete) usp_msg.Msg {
+	return usp_msg.Msg{
+		Header: &usp_msg.Header{
+			MsgId:   uuid.NewString(),
+			MsgType: usp_msg.Header_DELETE,
+		},
+		Body: &usp_msg.Body{
+			MsgBody: &usp_msg.Body_Request{
+				Request: &usp_msg.Request{
+					ReqType: &usp_msg.Request_Delete{
+						Delete: &getStuff,
+					},
+				},
+			},
+		},
+	}
+}
+
+func NewSetMsg(updateStuff usp_msg.Set) usp_msg.Msg {
+	return usp_msg.Msg{
+		Header: &usp_msg.Header{
+			MsgId:   uuid.NewString(),
+			MsgType: usp_msg.Header_SET,
+		},
+		Body: &usp_msg.Body{
+			MsgBody: &usp_msg.Body_Request{
+				Request: &usp_msg.Request{
+					ReqType: &usp_msg.Request_Set{
+						Set: &updateStuff,
+					},
+				},
+			},
+		},
+	}
+}
+
+func NewGetSupportedParametersMsg(getStuff usp_msg.GetSupportedDM) usp_msg.Msg {
+	return usp_msg.Msg{
+		Header: &usp_msg.Header{
+			MsgId:   uuid.NewString(),
+			MsgType: usp_msg.Header_GET_SUPPORTED_DM,
+		},
+		Body: &usp_msg.Body{
+			MsgBody: &usp_msg.Body_Request{
+				Request: &usp_msg.Request{
+					ReqType: &usp_msg.Request_GetSupportedDm{
+						GetSupportedDm: &getStuff,
+					},
+				},
+			},
+		},
+	}
+}
+
+func NewGetParametersInstancesMsg(getStuff usp_msg.GetInstances) usp_msg.Msg {
+	return usp_msg.Msg{
+		Header: &usp_msg.Header{
+			MsgId:   uuid.NewString(),
+			MsgType: usp_msg.Header_GET_INSTANCES,
+		},
+		Body: &usp_msg.Body{
+			MsgBody: &usp_msg.Body_Request{
+				Request: &usp_msg.Request{
+					ReqType: &usp_msg.Request_GetInstances{
+						GetInstances: &getStuff,
+					},
+				},
+			},
+		},
+	}
+}
+
+func NewOperateMsg(getStuff usp_msg.Operate) usp_msg.Msg {
+	return usp_msg.Msg{
+		Header: &usp_msg.Header{
+			MsgId:   uuid.NewString(),
+			MsgType: usp_msg.Header_OPERATE,
+		},
+		Body: &usp_msg.Body{
+			MsgBody: &usp_msg.Body_Request{
+				Request: &usp_msg.Request{
+					ReqType: &usp_msg.Request_Operate{
+						Operate: &getStuff,
+					},
+				},
+			},
+		},
+	}
+}

backend/services/mtp/mqtt-adapter/README.md

@@ -0,0 +1 @@
+- acts as a bridge between mqtt server and controller

backend/services/mtp/mqtt-adapter/cmd/mqtt-adapter/main.go

@@ -0,0 +1,29 @@
+package main
+
+import (
+	"log"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"github.com/OktopUSP/oktopus/backend/services/mqtt-adapter/internal/bridge"
+	"github.com/OktopUSP/oktopus/backend/services/mqtt-adapter/internal/config"
+	"github.com/OktopUSP/oktopus/backend/services/mqtt-adapter/internal/nats"
+)
+
+func main() {
+
+	done := make(chan os.Signal, 1)
+	signal.Notify(done, syscall.SIGINT, syscall.SIGTERM)
+
+	c := config.NewConfig()
+
+	_, publisher, subscriber := nats.StartNatsClient(c.Nats)
+
+	bridge := bridge.NewBridge(publisher, subscriber, c.Mqtt.Ctx, c.Mqtt)
+	bridge.StartBridge()
+
+	<-done
+
+	log.Println("mqtt adapter is shutting down...")
+}

backend/services/mtp/mqtt-adapter/go.mod

@@ -0,0 +1,18 @@
+module github.com/OktopUSP/oktopus/backend/services/mqtt-adapter
+
+go 1.22.0
+
+require (
+	github.com/eclipse/paho.golang v0.21.0 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/gorilla/websocket v1.5.1 // indirect
+	github.com/joho/godotenv v1.5.1 // indirect
+	github.com/klauspost/compress v1.17.7 // indirect
+	github.com/nats-io/nats.go v1.33.1 // indirect
+	github.com/nats-io/nkeys v0.4.7 // indirect
+	github.com/nats-io/nuid v1.0.1 // indirect
+	golang.org/x/crypto v0.20.0 // indirect
+	golang.org/x/net v0.21.0 // indirect
+	golang.org/x/sys v0.17.0 // indirect
+	golang.org/x/text v0.14.0 // indirect
+)

backend/services/mtp/mqtt-adapter/go.sum

@@ -0,0 +1,24 @@
+github.com/eclipse/paho.golang v0.21.0 h1:cxxEReu+iFbA5RrHfRGxJOh8tXZKDywuehneoeBeyn8=
+github.com/eclipse/paho.golang v0.21.0/go.mod h1:GHF6vy7SvDbDHBguaUpfuBkEB5G6j0zKxMG4gbh6QRQ=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/gorilla/websocket v1.5.1 h1:gmztn0JnHVt9JZquRuzLw3g4wouNVzKL15iLr/zn/QY=
+github.com/gorilla/websocket v1.5.1/go.mod h1:x3kM2JMyaluk02fnUJpQuwD2dCS5NDG2ZHL0uE0tcaY=
+github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=
+github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4=
+github.com/klauspost/compress v1.17.7 h1:ehO88t2UGzQK66LMdE8tibEd1ErmzZjNEqWkjLAKQQg=
+github.com/klauspost/compress v1.17.7/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
+github.com/nats-io/nats.go v1.33.1 h1:8TxLZZ/seeEfR97qV0/Bl939tpDnt2Z2fK3HkPypj70=
+github.com/nats-io/nats.go v1.33.1/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8=
+github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI=
+github.com/nats-io/nkeys v0.4.7/go.mod h1:kqXRgRDPlGy7nGaEDMuYzmiJCIAAWDK0IMBtDmGD0nc=
+github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
+github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
+golang.org/x/crypto v0.20.0 h1:jmAMJJZXr5KiCw05dfYK9QnqaqKLYXijU23lsEdcQqg=
+golang.org/x/crypto v0.20.0/go.mod h1:Xwo95rrVNIoSMx9wa1JroENMToLWn3RNVrTBpLHgZPQ=
+golang.org/x/net v0.21.0 h1:AQyQV4dYCvJ7vGmJyKki9+PBdyvhkSd8EIx/qb0AYv4=
+golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44=
+golang.org/x/sys v0.17.0 h1:25cE3gD+tdBA7lp7QfhuV+rJiE9YXTcS3VG1SqssI/Y=
+golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
+golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=

backend/services/mtp/mqtt-adapter/internal/bridge/bridge.go

@@ -0,0 +1,196 @@
+package bridge
+
+import (
+	"context"
+	"log"
+	"net/url"
+	"strings"
+	"time"
+
+	"github.com/OktopUSP/oktopus/backend/services/mqtt-adapter/internal/config"
+	"github.com/eclipse/paho.golang/autopaho"
+	"github.com/eclipse/paho.golang/paho"
+	"github.com/google/uuid"
+	"github.com/nats-io/nats.go"
+)
+
+const (
+	ONLINE = iota
+	OFFLINE
+)
+
+const NATS_MQTT_SUBJECT_PREFIX = "mqtt.usp.v1."
+const NATS_MQTT_ADAPTER_SUBJECT_PREFIX = "mqtt-adapter.usp.v1.*."
+const MQTT_TOPIC_PREFIX = "oktopus/usp/"
+
+type (
+	Publisher  func(string, []byte) error
+	Subscriber func(string, func(*nats.Msg)) error
+)
+
+type Bridge struct {
+	Pub  Publisher
+	Sub  Subscriber
+	Mqtt config.Mqtt
+	Ctx  context.Context
+}
+
+func NewBridge(p Publisher, s Subscriber, ctx context.Context, m config.Mqtt) *Bridge {
+	return &Bridge{
+		Pub:  p,
+		Sub:  s,
+		Mqtt: m,
+		Ctx:  ctx,
+	}
+}
+
+func (b *Bridge) StartBridge() {
+
+	broker, _ := url.Parse(b.Mqtt.Url)
+
+	status := make(chan *paho.Publish)
+	controller := make(chan *paho.Publish)
+	apiMsg := make(chan *paho.Publish)
+
+	go b.mqttMessageHandler(status, controller, apiMsg)
+
+	pahoClientConfig := buildClientConfig(status, controller, apiMsg, b.Mqtt.ClientId)
+
+	autopahoClientConfig := autopaho.ClientConfig{
+		BrokerUrls: []*url.URL{
+			broker,
+		},
+		KeepAlive:         30,
+		ConnectRetryDelay: 5 * time.Second,
+		ConnectTimeout:    5 * time.Second,
+		OnConnectionUp: func(cm *autopaho.ConnectionManager, connAck *paho.Connack) {
+			log.Printf("Connected to MQTT broker--> %s", b.Mqtt.Url)
+			subscribe(b.Mqtt.Ctx, b.Mqtt.Qos, cm)
+		},
+		OnConnectError: func(err error) {
+			log.Printf("Error while attempting connection: %s\n", err)
+		},
+		ClientConfig: *pahoClientConfig,
+	}
+
+	if b.Mqtt.Username != "" && b.Mqtt.Password != "" {
+		autopahoClientConfig.SetUsernamePassword(b.Mqtt.Username, []byte(b.Mqtt.Password))
+	}
+
+	log.Println("MQTT client id:", pahoClientConfig.ClientID)
+	log.Println("MQTT username:", b.Mqtt.Username)
+	log.Println("MQTT password:", b.Mqtt.Password)
+
+	cm, err := autopaho.NewConnection(b.Ctx, autopahoClientConfig)
+	if err != nil {
+		log.Fatalln(err)
+	}
+
+	b.natsMessageHandler(cm)
+}
+
+func (b *Bridge) natsMessageHandler(cm *autopaho.ConnectionManager) {
+	b.Sub(NATS_MQTT_ADAPTER_SUBJECT_PREFIX+"info", func(m *nats.Msg) {
+
+		log.Printf("Received message on info subject")
+		cm.Publish(b.Ctx, &paho.Publish{
+			QoS:     byte(b.Mqtt.Qos),
+			Topic:   MQTT_TOPIC_PREFIX + "v1/agent/" + getDeviceFromSubject(m.Subject),
+			Payload: m.Data,
+			Properties: &paho.PublishProperties{
+				ResponseTopic: "oktopus/usp/v1/controller/" + getDeviceFromSubject(m.Subject),
+			},
+		})
+	})
+}
+
+func getDeviceFromSubject(subject string) string {
+	paths := strings.Split(subject, ".")
+	device := paths[len(paths)-2]
+	return device
+}
+
+func (b *Bridge) mqttMessageHandler(status, controller, apiMsg chan *paho.Publish) {
+	for {
+		select {
+		case d := <-status:
+			b.Pub(NATS_MQTT_SUBJECT_PREFIX+getDeviceFromTopic(d.Topic)+".status", d.Payload)
+		case c := <-controller:
+			b.Pub(NATS_MQTT_SUBJECT_PREFIX+getDeviceFromTopic(c.Topic)+".info", c.Payload)
+		case a := <-apiMsg:
+			b.Pub(NATS_MQTT_SUBJECT_PREFIX+getDeviceFromTopic(a.Topic)+".api", a.Payload)
+		}
+	}
+}
+
+func getDeviceFromTopic(topic string) string {
+	paths := strings.Split(topic, "/")
+	device := paths[len(paths)-1]
+	return device
+}
+
+func subscribe(ctx context.Context, qos int, c *autopaho.ConnectionManager) {
+	if _, err := c.Subscribe(ctx, &paho.Subscribe{
+		Subscriptions: []paho.SubscribeOptions{
+			{
+				Topic: MQTT_TOPIC_PREFIX + "+/api/+",
+				QoS:   byte(qos),
+			},
+			{
+				Topic: MQTT_TOPIC_PREFIX + "+/controller/+",
+				QoS:   byte(qos),
+			},
+			{
+				Topic: MQTT_TOPIC_PREFIX + "+/status/+",
+				QoS:   byte(qos),
+			},
+		},
+	}); err != nil {
+		log.Fatalln(err)
+	}
+
+	log.Printf("Subscribed to %s", MQTT_TOPIC_PREFIX+"+/controller/+")
+	log.Printf("Subscribed to %s", MQTT_TOPIC_PREFIX+"+/status/+")
+	log.Printf("Subscribed to %s", MQTT_TOPIC_PREFIX+"+/api/+")
+}
+
+func buildClientConfig(status, controller, apiMsg chan *paho.Publish, id string) *paho.ClientConfig {
+	log.Println("Starting new MQTT client")
+	singleHandler := paho.NewSingleHandlerRouter(func(p *paho.Publish) {
+
+		if strings.Contains(p.Topic, "status") {
+			status <- p
+		} else if strings.Contains(p.Topic, "controller") {
+			controller <- p
+		} else if strings.Contains(p.Topic, "api") {
+			apiMsg <- p
+		} else {
+			log.Println("No handler for topic: ", p.Topic)
+		}
+
+	})
+
+	clientConfig := paho.ClientConfig{}
+
+	clientConfig = paho.ClientConfig{
+		Router: singleHandler,
+		OnServerDisconnect: func(d *paho.Disconnect) {
+			if d.Properties != nil {
+				log.Printf("Requested disconnect: %s\n , properties reason: %s\n", clientConfig.ClientID, d.Properties.ReasonString)
+			} else {
+				log.Printf("Requested disconnect; %s reason code: %d\n", clientConfig.ClientID, d.ReasonCode)
+			}
+		},
+		OnClientError: func(err error) {
+			log.Println(err)
+		},
+	}
+
+	if id != "" {
+		clientConfig.ClientID = id
+	} else {
+		clientConfig.ClientID = uuid.NewString()
+	}
+
+	return &clientConfig
+}

backend/services/mtp/mqtt-adapter/internal/config/config.go

@@ -0,0 +1,127 @@
+package config
+
+import (
+	"context"
+	"flag"
+	"log"
+	"os"
+	"strconv"
+
+	"github.com/joho/godotenv"
+)
+
+const LOCAL_ENV = ".env.local"
+
+type Nats struct {
+	Url                string
+	Name               string
+	VerifyCertificates bool
+	Ctx                context.Context
+}
+
+type Mqtt struct {
+	Url      string
+	ClientId string
+	Username string
+	Password string
+	Qos      int
+	Ctx      context.Context
+}
+
+type Config struct {
+	Nats Nats
+	Mqtt Mqtt
+}
+
+func NewConfig() *Config {
+
+	loadEnvVariables()
+	log.SetFlags(log.LstdFlags | log.Lshortfile)
+
+	natsUrl := flag.String("nats_url", lookupEnvOrString("NATS_URL", "nats://localhost:4222"), "url for nats server")
+	natsName := flag.String("nats_name", lookupEnvOrString("NATS_NAME", "mqtt-adapter"), "name for nats client")
+	natsVerifyCertificates := flag.Bool("nats_verify_certificates", lookupEnvOrBool("NATS_VERIFY_CERTIFICATES", false), "verify validity of certificates from nats server")
+	mqttUrl := flag.String("mqtt_url", lookupEnvOrString("MQTT_URL", "tcp://localhost:1883"), "url for mqtt server")
+	mqttClientId := flag.String("mqtt_client_id", lookupEnvOrString("MQTT_CLIENT_ID", "mqtt-adapter"), "client id for mqtt")
+	mqttUsername := flag.String("mqtt_username", lookupEnvOrString("MQTT_USERNAME", ""), "username for mqtt")
+	mqttPassword := flag.String("mqtt_password", lookupEnvOrString("MQTT_PASSWORD", ""), "password for mqtt")
+	mqttQos := flag.Int("mqtt_qos", lookupEnvOrInt("MQTT_QOS", 1), "quality of service for mqtt")
+	flHelp := flag.Bool("help", false, "Help")
+
+	/*
+		App variables priority:
+		1º - Flag through command line.
+		2º - Env variables.
+		3º - Default flag value.
+	*/
+
+	flag.Parse()
+
+	if *flHelp {
+		flag.Usage()
+		os.Exit(0)
+	}
+
+	ctx := context.TODO()
+
+	return &Config{
+		Nats: Nats{
+			Url:                *natsUrl,
+			Name:               *natsName,
+			VerifyCertificates: *natsVerifyCertificates,
+			Ctx:                ctx,
+		},
+		Mqtt: Mqtt{
+			Url:      *mqttUrl,
+			ClientId: *mqttClientId,
+			Username: *mqttUsername,
+			Password: *mqttPassword,
+			Ctx:      ctx,
+			Qos:      *mqttQos,
+		},
+	}
+}
+
+func loadEnvVariables() {
+	err := godotenv.Load()
+
+	if _, err := os.Stat(LOCAL_ENV); err == nil {
+		_ = godotenv.Overload(LOCAL_ENV)
+		log.Printf("Loaded variables from '%s'", LOCAL_ENV)
+	}
+
+	if err != nil {
+		log.Println("Error to load environment variables:", err)
+	} else {
+		log.Println("Loaded variables from '.env'")
+	}
+}
+
+func lookupEnvOrString(key string, defaultVal string) string {
+	if val, _ := os.LookupEnv(key); val != "" {
+		return val
+	}
+	return defaultVal
+}
+
+func lookupEnvOrInt(key string, defaultVal int) int {
+	if val, _ := os.LookupEnv(key); val != "" {
+		v, err := strconv.Atoi(val)
+		if err != nil {
+			log.Fatalf("LookupEnvOrInt[%s]: %v", key, err)
+		}
+		return v
+	}
+	return defaultVal
+}
+
+func lookupEnvOrBool(key string, defaultVal bool) bool {
+	if val, _ := os.LookupEnv(key); val != "" {
+		v, err := strconv.ParseBool(val)
+		if err != nil {
+			log.Fatalf("LookupEnvOrBool[%s]: %v", key, err)
+		}
+		return v
+	}
+	return defaultVal
+}

backend/services/mtp/mqtt-adapter/internal/nats/nats.go

@@ -0,0 +1,92 @@
+package nats
+
+import (
+	"log"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"github.com/OktopUSP/oktopus/backend/services/mqtt-adapter/internal/config"
+)
+
+const (
+	STREAM_NAME = "mqtt"
+)
+
+func StartNatsClient(c config.Nats) (
+	*nats.Conn,
+	func(string, []byte) error,
+	func(string, func(*nats.Msg)) error,
+) {
+
+	var (
+		nc  *nats.Conn
+		err error
+	)
+
+	opts := defineOptions(c)
+
+	for {
+		nc, err = nats.Connect(c.Url, opts...)
+		if err != nil {
+			time.Sleep(5 * time.Second)
+			continue
+		}
+		break
+	}
+	log.Printf("Successfully connected to NATS server %s", c.Url)
+
+	js, err := jetstream.New(nc)
+	if err != nil {
+		log.Fatalf("Failed to create JetStream client: %v", err)
+	}
+
+	nc.Subscribe("opa.123.dae", func(m *nats.Msg) {
+		log.Printf("Received message on subject %s: %s", m.Subject, string(m.Data))
+	})
+
+	return nc, publisher(js), subscriber(nc)
+}
+
+func subscriber(nc *nats.Conn) func(string, func(*nats.Msg)) error {
+	return func(subject string, handler func(*nats.Msg)) error {
+		_, err := nc.Subscribe(subject, handler)
+		if err != nil {
+			log.Printf("error to subscribe to subject %s error: %q", subject, err)
+		}
+		return err
+	}
+}
+
+func publisher(js jetstream.JetStream) func(string, []byte) error {
+	return func(subject string, payload []byte) error {
+		_, err := js.PublishAsync(subject, payload)
+		if err != nil {
+			log.Printf("error to send jetstream message: %q", err)
+		}
+		return err
+	}
+}
+
+func defineOptions(c config.Nats) []nats.Option {
+	var opts []nats.Option
+
+	opts = append(opts, nats.Name(c.Name))
+	opts = append(opts, nats.MaxReconnects(-1))
+	opts = append(opts, nats.ReconnectWait(5*time.Second))
+	opts = append(opts, nats.DisconnectErrHandler(func(nc *nats.Conn, err error) {
+		log.Printf("Got disconnected! Reason: %q\n", err)
+	}))
+	opts = append(opts, nats.ReconnectHandler(func(nc *nats.Conn) {
+		log.Printf("Got reconnected to %v!\n", nc.ConnectedUrl())
+	}))
+	opts = append(opts, nats.ClosedHandler(func(nc *nats.Conn) {
+		log.Printf("Connection closed. Reason: %q\n", nc.LastError())
+	}))
+	if c.VerifyCertificates {
+		opts = append(opts, nats.RootCAs())
+	}
+
+	return opts
+}

backend/services/mtp/mqtt/internal/listeners/mqtt/hook.go

@@ -42,7 +42,7 @@ func (h *MyHook) OnDisconnect(cl *mqtt.Client, err error, expire bool) {
 	}
 
 	if clUser != "" {
-		err := server.Publish("oktopus/v1/status/"+clUser, []byte("1"), false, 1)
+		err := server.Publish("oktopus/usp/v1/status/"+clUser, []byte("1"), false, 1)
 		if err != nil {
 			log.Println("server publish error: ", err)
 		}
@@ -51,7 +51,7 @@ func (h *MyHook) OnDisconnect(cl *mqtt.Client, err error, expire bool) {
 
 func (h *MyHook) OnSubscribed(cl *mqtt.Client, pk packets.Packet, reasonCodes []byte) {
 	//Verifies if it's a device who is subscribed
-	if strings.Contains(pk.Filters[0].Filter, "oktopus/v1/agent") {
+	if strings.Contains(pk.Filters[0].Filter, "oktopus/usp/v1/agent") {
 		var clUser string
 
 		if len(cl.Properties.Props.User) > 0 {
@@ -61,12 +61,12 @@ func (h *MyHook) OnSubscribed(cl *mqtt.Client, pk packets.Packet, reasonCodes []
 		if clUser != "" {
 			cl.Properties.Will = mqtt.Will{
 				Qos:       1,
-				TopicName: "oktopus/v1/status/" + clUser,
+				TopicName: "oktopus/usp/v1/status/" + clUser,
 				Payload:   []byte("1"),
 				Retain:    false,
 			}
 			log.Println("new device:", clUser)
-			err := server.Publish("oktopus/v1/status/"+clUser, []byte("0"), false, 1)
+			err := server.Publish("oktopus/usp/v1/status/"+clUser, []byte("0"), false, 1)
 			if err != nil {
 				log.Println("server publish error: ", err)
 			}
@@ -81,7 +81,7 @@ func (h *MyHook) OnPacketEncode(cl *mqtt.Client, pk packets.Packet) packets.Pack
 		clUser = cl.Properties.Props.User[0].Val
 	}
 	if pk.FixedHeader.Type == packets.Connack {
-		pk.Properties.User = []packets.UserProperty{{Key: "subscribe-topic", Val: "oktopus/v1/agent/" + clUser}}
+		pk.Properties.User = []packets.UserProperty{{Key: "subscribe-topic", Val: "oktopus/usp/v1/agent/" + clUser}}
 	}
 
 	return pk

backend/services/mtp/mqtt/internal/listeners/mqtt/mqtt.go

@@ -35,18 +35,20 @@ type Redis struct {
 	RedisPassword string
 }
 
-func (m *Mqtt) Start(server *mqtt.Server) {
+func (m *Mqtt) Start(mqttServer *mqtt.Server) {
 
-	defineSeverLog(server, m.LogLevel)
+	defineSeverLog(mqttServer, m.LogLevel)
-	defineServerAuth(server, m.AuthFile)
+	defineServerAuth(mqttServer, m.AuthFile)
+
+	server = mqttServer
 
 	var tlsConfig *listeners.Config
 	if m.Tls {
 		tlsConfig = defineServerTls(m.Fullchain, m.Privkey)
 	}
 
-	createListener(server, m.Port, tlsConfig)
+	createListener(mqttServer, m.Port, tlsConfig)
-	addHooks(server, m.Redis)
+	addHooks(mqttServer, m.Redis)
 }
 
 func addHooks(server *mqtt.Server, redisConf Redis) {