text: package mqtt
  
  import (
  	"bytes"
  	"context"
  	"crypto/tls"
  	"crypto/x509"
  	"encoding/base64"
  	"encoding/json"
  	"fmt"
  	"io/ioutil"
  	"regexp"
  	"strconv"
  	"sync"
  	"text/template"
  	"time"
  
  	mqtt "github.com/eclipse/paho.mqtt.golang"
  	"github.com/golang/protobuf/proto"
  	"github.com/pkg/errors"
  	log "github.com/sirupsen/logrus"
  
  	pb "github.com/brocaar/chirpstack-api/go/v3/as/integration"
  	"github.com/brocaar/chirpstack-application-server/internal/config"
  	"github.com/brocaar/chirpstack-application-server/internal/integration/marshaler"
  	"github.com/brocaar/chirpstack-application-server/internal/integration/models"
  	"github.com/brocaar/chirpstack-application-server/internal/logging"
  	"github.com/brocaar/chirpstack-application-server/internal/storage"
  	"github.com/brocaar/lorawan"
  )
  
  const downlinkLockTTL = time.Millisecond * 100
  
  // Integration implements a MQTT integration.
  type Integration struct {
  	marshaler            marshaler.Type
  	conn                 mqtt.Client
  	dataDownChan         chan models.DataDownPayload
  	wg                   sync.WaitGroup
  	config               config.IntegrationMQTTConfig
  	eventTopicTemplate   *template.Template
  	commandTopicTemplate *template.Template
  	downlinkTopic        string
  	downlinkRegexp       *regexp.Regexp
  	retainEvents         bool
  
  	// For backwards compatibility.
  	uplinkTemplate      *template.Template
  	downlinkTemplate    *template.Template
  	joinTemplate        *template.Template
  	ackTemplate         *template.Template
  	errorTemplate       *template.Template
  	statusTemplate      *template.Template
  	locationTemplate    *template.Template
  	txAckTemplate       *template.Template
  	integrationTemplate *template.Template
  	uplinkRetained      bool
  	joinRetained        bool
  	ackRetained         bool
  	errorRetained       bool
  	statusRetained      bool
  	locationRetained    bool
  	txAckRetained       bool
  	integrationRetained bool
  }
  
  // New creates a new MQTT integration.
  func New(m marshaler.Type, conf config.IntegrationMQTTConfig) (*Integration, error) {
  	var err error
  	i := Integration{
  		marshaler:    m,
  		dataDownChan: make(chan models.DataDownPayload),
  		config:       conf,
  	}
  
  	i.retainEvents = i.config.RetainEvents
  	i.eventTopicTemplate, err = template.New("event").Parse(i.config.EventTopicTemplate)
  	if err != nil {
  		return nil, errors.Wrap(err, "parse event template error")
  	}
  	i.commandTopicTemplate, err = template.New("command").Parse(i.config.CommandTopicTemplate)
  	if err != nil {
  		return nil, errors.Wrap(err, "parse command template error")
  	}
  
  	// For backwards compatibility.
  	if i.config.UplinkTopicTemplate != "" {
  		i.uplinkTemplate, err = template.New("uplink").Parse(i.config.UplinkTopicTemplate)
  		if err != nil {
  			return nil, errors.Wrap(err, "parse uplink template error")
  		}
  	}
  	if i.config.DownlinkTopicTemplate != "" {
  		i.downlinkTemplate, err = template.New("downlink").Parse(i.config.DownlinkTopicTemplate)
  		if err != nil {
  			return nil, errors.Wrap(err, "parse downlink template error")
  		}
  	}
  	if i.config.JoinTopicTemplate != "" {
  		i.joinTemplate, err = template.New("join").Parse(i.config.JoinTopicTemplate)
  		if err != nil {
  			return nil, errors.Wrap(err, "parse join template error")
  		}
  	}
  	if i.config.AckTopicTemplate != "" {
  		i.ackTemplate, err = template.New("ack").Parse(i.config.AckTopicTemplate)
  		if err != nil {
  			return nil, errors.Wrap(err, "parse ack template error")
  		}
  	}
  	if i.config.ErrorTopicTemplate != "" {
  		i.errorTemplate, err = template.New("error").Parse(i.config.ErrorTopicTemplate)
  		if err != nil {
  			return nil, errors.Wrap(err, "parse error template error")
  		}
  	}
  	if i.config.StatusTopicTemplate != "" {
  		i.statusTemplate, err = template.New("status").Parse(i.config.StatusTopicTemplate)
  		if err != nil {
  			return nil, errors.Wrap(err, "parse status template error")
  		}
  	}
  	if i.config.LocationTopicTemplate != "" {
  		i.locationTemplate, err = template.New("location").Parse(i.config.LocationTopicTemplate)
  		if err != nil {
  			return nil, errors.Wrap(err, "parse location template error")
  		}
  	}
  	if i.config.TxAckTopicTemplate != "" {
  		i.txAckTemplate, err = template.New("txack").Parse(i.config.TxAckTopicTemplate)
  		if err != nil {
  			return nil, errors.Wrap(err, "parse tx ack template error")
  		}
  	}
  	if i.config.IntegrationTopicTemplate != "" {
  		i.integrationTemplate, err = template.New("integration").Parse(i.config.IntegrationTopicTemplate)
  		if err != nil {
  			return nil, errors.Wrap(err, "parse integration template error")
  		}
  	}
  	i.uplinkRetained = i.config.UplinkRetainedMessage
  	i.joinRetained = i.config.JoinRetainedMessage
  	i.ackRetained = i.config.AckRetainedMessage
  	i.errorRetained = i.config.ErrorRetainedMessage
  	i.statusRetained = i.config.StatusRetainedMessage
  	i.locationRetained = i.config.LocationRetainedMessage
  	i.txAckRetained = i.config.TxAckRetainedMessage
  	i.integrationRetained = i.config.IntegrationRetainedMessage
  
  	// generate downlink topic matching all applications and devices
  	i.downlinkTopic, err = i.getDownlinkTopic()
  	if err != nil {
  		return nil, errors.Wrap(err, "get downlink topic error")
  	}
  
  	// generate downlink topic regexp
  	i.downlinkRegexp, err = i.getDownlinkTopicRegexp()
  	if err != nil {
  		return nil, errors.Wrap(err, "get downlink topic regexp error")
  	}
  
  	opts := mqtt.NewClientOptions()
  	opts.AddBroker(i.config.Server)
  	opts.SetUsername(i.config.Username)
  	opts.SetPassword(i.config.Password)
  	opts.SetCleanSession(i.config.CleanSession)
  	opts.SetClientID(i.config.ClientID)
  	opts.SetOnConnectHandler(i.onConnected)
  	opts.SetConnectionLostHandler(i.onConnectionLost)
  	opts.SetMaxReconnectInterval(i.config.MaxReconnectInterval)
  
  	tlsconfig, err := newTLSConfig(i.config.CACert, i.config.TLSCert, i.config.TLSKey)
  	if err != nil {
  		log.WithError(err).WithFields(log.Fields{
  			"ca_cert":  i.config.CACert,
  			"tls_cert": i.config.TLSCert,
  			"tls_key":  i.config.TLSKey,
  		}).Fatalf("error loading mqtt certificate files")
  	}
  	if tlsconfig != nil {
  		opts.SetTLSConfig(tlsconfig)
  	}
  
  	log.WithField("server", i.config.Server).Info("integration/mqtt: connecting to mqtt broker")
  	i.conn = mqtt.NewClient(opts)
  	for {
  		if token := i.conn.Connect(); token.Wait() && token.Error() != nil {
  			log.Errorf("integration/mqtt: connecting to broker error, will retry in 2s: %s", token.Error())
  			time.Sleep(2 * time.Second)
  		} else {
  			break
  		}
  	}
  	return &i, nil
  }
  
  func newTLSConfig(cafile, certFile, certKeyFile string) (*tls.Config, error) {
  	// Here are three valid options:
  	//   - Only CA
  	//   - TLS cert + key
  	//   - CA, TLS cert + key
  
  	if cafile == "" && certFile == "" && certKeyFile == "" {
  		log.Info("integration/mqtt: TLS config is empty")
  		return nil, nil
  	}
  
  	tlsConfig := &tls.Config{}
  
  	// Import trusted certificates from CAfile.pem.
  	if cafile != "" {
  		cacert, err := ioutil.ReadFile(cafile)
  		if err != nil {
  			log.Errorf("integration/mqtt: couldn't load cafile: %s", err)
  			return nil, err
  		}
  		certpool := x509.NewCertPool()
  		certpool.AppendCertsFromPEM(cacert)
  
  		tlsConfig.RootCAs = certpool // RootCAs = certs used to verify server cert.
  	}
  
  	// Import certificate and the key
  	if certFile != "" || certKeyFile != "" {
  		kp, err := tls.LoadX509KeyPair(certFile, certKeyFile) // here raises error when the pair of cert and key are invalid (e.g. either one is empty)
  		if err != nil {
  			log.Errorf("integration/mqtt: couldn't load MQTT TLS key pair: %s", err)
  			return nil, err
  		}
  		tlsConfig.Certificates = []tls.Certificate{kp}
  	}
  
  	return tlsConfig, nil
  }
  
  // Close stops the handler.
  func (i *Integration) Close() error {
  	log.Info("integration/mqtt: closing handler")
  	log.WithField("topic", i.downlinkTopic).Info("integration/mqtt: unsubscribing from tx topic")
  	if token := i.conn.Unsubscribe(i.downlinkTopic); token.Wait() && token.Error() != nil {
  		return fmt.Errorf("integration/mqtt: unsubscribe from %s error: %s", i.downlinkTopic, token.Error())
  	}
  	log.Info("integration/mqtt: handling last items in queue")
  	i.wg.Wait()
  	close(i.dataDownChan)
  	return nil
  }
  
  // HandleUplinkEvent sends an UplinkEvent.
  func (i *Integration) HandleUplinkEvent(ctx context.Context, _ models.Integration, vars map[string]string, payload pb.UplinkEvent) error {
  	return i.publish(ctx, payload.ApplicationId, payload.DevEui, "up", &payload)
  }
  
  // HandleJoinEvent sends a JoinEvent.
  func (i *Integration) HandleJoinEvent(ctx context.Context, _ models.Integration, vars map[string]string, payload pb.JoinEvent) error {
  	return i.publish(ctx, payload.ApplicationId, payload.DevEui, "join", &payload)
  }
  
  // HandleAckEvent sends an AckEvent.
  func (i *Integration) HandleAckEvent(ctx context.Context, _ models.Integration, vars map[string]string, payload pb.AckEvent) error {
  	return i.publish(ctx, payload.ApplicationId, payload.DevEui, "ack", &payload)
  }
  
  // HandleErrorEvent sends an ErrorEvent.
  func (i *Integration) HandleErrorEvent(ctx context.Context, _ models.Integration, vars map[string]string, payload pb.ErrorEvent) error {
  	return i.publish(ctx, payload.ApplicationId, payload.DevEui, "error", &payload)
  }
  
  // HandleStatusEvent sends a StatusEvent.
  func (i *Integration) HandleStatusEvent(ctx context.Context, _ models.Integration, vars map[string]string, payload pb.StatusEvent) error {
  	return i.publish(ctx, payload.ApplicationId, payload.DevEui, "status", &payload)
  }
  
  // HandleLocationEvent sends a LocationEvent.
  func (i *Integration) HandleLocationEvent(ctx context.Context, _ models.Integration, vars map[string]string, payload pb.LocationEvent) error {
  	return i.publish(ctx, payload.ApplicationId, payload.DevEui, "location", &payload)
  }
  
  // HandleTxAckEvent sends a TxAckEvent.
  func (i *Integration) HandleTxAckEvent(ctx context.Context, _ models.Integration, vars map[string]string, payload pb.TxAckEvent) error {
  	return i.publish(ctx, payload.ApplicationId, payload.DevEui, "txack", &payload)
  }
  
  // HandleIntegrationEvent sends an IntegrationEvent.
  func (i *Integration) HandleIntegrationEvent(ctx context.Context, _ models.Integration, vars map[string]string, payload pb.IntegrationEvent) error {
  	return i.publish(ctx, payload.ApplicationId, payload.DevEui, "integration", &payload)
  }
  
  func (i *Integration) publish(ctx context.Context, applicationID uint64, devEUIB []byte, eventType string, msg proto.Message) error {
  	var devEUI lorawan.EUI64
  	copy(devEUI[:], devEUIB)
  
  	topic, err := i.getTopic(applicationID, devEUI, eventType)
  	if err != nil {
  		return errors.Wrap(err, "get topic error")
  	}
  
  	retain := i.getRetainEvents(eventType)
  
  	b, err := marshaler.Marshal(i.marshaler, msg)
  	if err != nil {
  		return err
  	}
  
  	log.WithFields(log.Fields{
  		"dev_eui": devEUI,
  		"retain":  retain,
  		"topic":   topic,
  		"qos":     i.config.QOS,
  		"ctx_id":  ctx.Value(logging.ContextIDKey),
  	}).Info("integration/mqtt: publishing event")
  	if token := i.conn.Publish(topic, i.config.QOS, retain, b); token.Wait() && token.Error() != nil {
  		return token.Error()
  	}
  
  	mqttEventCounter(eventType).Inc()
  
  	return nil
  }
  
  // DataDownChan returns the channel containing the received DataDownPayload.
  func (i *Integration) DataDownChan() chan models.DataDownPayload {
  	return i.dataDownChan
  }
  
  func (i *Integration) getTXTopicVariables(topic string) (int64, lorawan.EUI64, error) {
  	var applicationID int64
  	var devEUI lorawan.EUI64
  	var err error
  
  	match := i.downlinkRegexp.FindStringSubmatch(topic)
  	if len(match) != len(i.downlinkRegexp.SubexpNames()) {
  		return applicationID, devEUI, errors.New("topic regex match error")
  	}
  
  	result := make(map[string]string)
  	for i, name := range i.downlinkRegexp.SubexpNames() {
  		if i != 0 && name != "" {
  			result[name] = match[i]
  		}
  	}
  
  	if idStr, ok := result["application_id"]; ok {
  		applicationID, err = strconv.ParseInt(idStr, 10, 64)
  		if err != nil {
  			return applicationID, devEUI, errors.Wrap(err, "parse application id error")
  		}
  	} else {
  		return applicationID, devEUI, errors.New("topic regexp does not contain application id")
  	}
  
  	if devEUIStr, ok := result["dev_eui"]; ok {
  		if err = devEUI.UnmarshalText([]byte(devEUIStr)); err != nil {
  			return applicationID, devEUI, errors.Wrap(err, "parse deveui error")
  		}
  	}
  
  	return applicationID, devEUI, nil
  }
  
  func (i *Integration) txPayloadHandler(mqttc mqtt.Client, msg mqtt.Message) {
  	i.wg.Add(1)
  	defer i.wg.Done()
  
  	log.WithField("topic", msg.Topic()).Info("integration/mqtt: downlink event received")
  	topicApplicationID, topicDevEUI, err := i.getTXTopicVariables(msg.Topic())
  	if err != nil {
  		log.WithError(err).Warning("integration/mqtt: get variables from topic error")
  		return
  	}
  
  	var pl models.DataDownPayload
  	dec := json.NewDecoder(bytes.NewReader(msg.Payload()))
  	if err := dec.Decode(&pl); err != nil {
  		log.WithFields(log.Fields{
  			"data_base64": base64.StdEncoding.EncodeToString(msg.Payload()),
  		}).Errorf("integration/mqtt: tx payload unmarshal error: %s", err)
  		return
  	}
  
  	pl.ApplicationID = topicApplicationID
  	pl.DevEUI = topicDevEUI
  
  	if pl.FPort == 0 || pl.FPort > 224 {
  		log.WithFields(log.Fields{
  			"topic":   msg.Topic(),
  			"dev_eui": pl.DevEUI,
  			"f_port":  pl.FPort,
  		}).Error("integration/mqtt: fPort must be between 1 - 224")
  		return
  	}
  
  	// Since with MQTT all subscribers will receive the downlink messages sent
  	// by the application, the first instance receiving the message must lock it,
  	// so that other instances can ignore the message.
  	key := fmt.Sprintf("lora:as:downlink:lock:%d:%s", pl.ApplicationID, pl.DevEUI)
  	set, err := storage.RedisClient().SetNX(key, "lock", downlinkLockTTL).Result()
  	if err != nil {
  		log.WithError(err).Error("integration/mqtt: acquire lock error")
  		return
  	}
  
  	// If we could not set, it means it is already locked by an other process.
  	if !set {
  		return
  	}
  
  	mqttCommandCounter("down").Inc()
  
  	i.dataDownChan <- pl
  }
  
  func (i *Integration) onConnected(mqttc mqtt.Client) {
  	log.Info("integration/mqtt: connected to mqtt broker")
  	for {
  		log.WithFields(log.Fields{
  			"topic": i.downlinkTopic,
  			"qos":   i.config.QOS,
  		}).Info("integration/mqtt: subscribing to tx topic")
  		if token := i.conn.Subscribe(i.downlinkTopic, i.config.QOS, i.txPayloadHandler); token.Wait() && token.Error() != nil {
  			log.WithField("topic", i.downlinkTopic).Errorf("integration/mqtt: subscribe error: %s", token.Error())
  			time.Sleep(time.Second)
  			continue
  		}
  		return
  	}
  }
  
  func (i *Integration) onConnectionLost(mqttc mqtt.Client, reason error) {
  	log.Errorf("integration/mqtt: mqtt connection error: %s", reason)
  }
  
  func (i *Integration) getDownlinkTopic() (string, error) {
  	topic := bytes.NewBuffer(nil)
  	topicTemplate := i.commandTopicTemplate
  	if i.downlinkTemplate != nil {
  		topicTemplate = i.downlinkTemplate
  	}
  
  	err := topicTemplate.Execute(topic, struct {
  		ApplicationID string
  		DevEUI        string
  		CommandType   string
  	}{"+", "+", "down"})
  	if err != nil {
  		return "", errors.Wrap(err, "execute template error")
  	}
  	return topic.String(), nil
  }
  
  func (i *Integration) getDownlinkTopicRegexp() (*regexp.Regexp, error) {
  	topic := bytes.NewBuffer(nil)
  	topicTemplate := i.commandTopicTemplate
  	if i.downlinkTemplate != nil {
  		topicTemplate = i.downlinkTemplate
  	}
  
  	err := topicTemplate.Execute(topic, struct {
  		ApplicationID string
  		DevEUI        string
  		CommandType   string
  	}{`(?P<application_id>\w+)`, `(?P<dev_eui>\w+)`, `(?P<command_type>\w)`})
  	if err != nil {
  		return nil, errors.Wrap(err, "execute template error")
  	}
  
  	r, err := regexp.Compile(topic.String())
  	if err != nil {
  		return nil, errors.Wrap(err, "compile regexp error")
  	}
  
  	return r, nil
  }
  
  func (i *Integration) getTopic(applicationID uint64, devEUI lorawan.EUI64, eventType string) (string, error) {
  	var topicTemplate *template.Template
  
  	// For backwards compatibility.
  	switch eventType {
  	case "up":
  		topicTemplate = i.uplinkTemplate
  	case "join":
  		topicTemplate = i.joinTemplate
  	case "ack":
  		topicTemplate = i.ackTemplate
  	case "error":
  		topicTemplate = i.errorTemplate
  	case "status":
  		topicTemplate = i.statusTemplate
  	case "location":
  		topicTemplate = i.locationTemplate
  	case "txack":
  		topicTemplate = i.txAckTemplate
  	case "integration":
  		topicTemplate = i.integrationTemplate
  	}
  
  	if topicTemplate == nil {
  		topicTemplate = i.eventTopicTemplate
  	}
  
  	topic := bytes.NewBuffer(nil)
  	err := topicTemplate.Execute(topic, struct {
  		ApplicationID uint64
  		DevEUI        lorawan.EUI64
  		EventType     string
  	}{applicationID, devEUI, eventType})
  	if err != nil {
  		return "", errors.Wrap(err, "execute template error")
  	}
  
  	return topic.String(), nil
  }
  
  func (i *Integration) getRetainEvents(eventType string) bool {
  	if i.retainEvents {
  		return true
  	}
  
  	// For backwards compatibility
  	switch eventType {
  	case "up":
  		return i.uplinkRetained
  	case "join":
  		return i.joinRetained
  	case "ack":
  		return i.ackRetained
  	case "error":
  		return i.errorRetained
  	case "status":
  		return i.statusRetained
  	case "location":
  		return i.locationRetained
  	case "txack":
  		return i.txAckRetained
  	case "integration":
  		return i.integrationRetained
  	default:
  		return false
  	}
  }
  