func (as *ApplicationServer) matchSession(ctx context.Context, ids ttnpb.EndDeviceIdentifiers, dev *ttnpb.EndDevice, link *ttnpb.ApplicationLink, sessionKeyID []byte) ([]string, error) {
	logger := log.FromContext(ctx)
	var mask []string
	switch {
	case dev.Session != nil && bytes.Equal(dev.Session.SessionKeyId, sessionKeyID):
	case dev.PendingSession != nil && bytes.Equal(dev.PendingSession.SessionKeyId, sessionKeyID):
		dev.Session = dev.PendingSession
		dev.PendingSession = nil
		mask = ttnpb.AddFields(mask, "session", "pending_session")
		logger.Debug("Switched to pending session")
	default:
		appSKey, err := as.fetchAppSKey(ctx, ids, sessionKeyID)
		if err != nil {
			return nil, errFetchAppSKey.WithCause(err)
		}
		dev.Session = &ttnpb.Session{
			DevAddr: *ids.DevAddr,
			SessionKeys: ttnpb.SessionKeys{
				SessionKeyId: sessionKeyID,
				AppSKey:      &appSKey,
			},
			StartedAt: time.Now().UTC(),
		}
		dev.PendingSession = nil
		dev.DevAddr = ids.DevAddr
		mask = ttnpb.AddFields(mask, "session", "pending_session", "ids.dev_addr")
		logger.Debug("Restored session")
	}
	return mask, nil
}
matchSession updates the currently active ttnpb.Session of a ttnpb.EndDevice, based on the provided session key ID.
This function will mutate the provided ttnpb.EndDevice and migrate the Session field to the session that matches
the provided session key ID.
The following fields are expected to be part of the provided ttnpb.EndDevice:
- session and pending_session - used to decide which session is currently active.
- formatters, version_ids - used by the downlink queue encoders, in cases in which the queue must be recalculated.
- skip_payload_crypto_override - used by the downlink queue migration mechanism in order to avoid payload encryption.
package applicationserver
import (
	"bytes"
	"context"
	"crypto/tls"
	"fmt"
	"net"
	"sync"
	"time"
	pbtypes "github.com/gogo/protobuf/types"
	"github.com/grpc-ecosystem/grpc-gateway/runtime"
	"go.thethings.network/lorawan-stack/v3/pkg/applicationserver/distribution"
	"go.thethings.network/lorawan-stack/v3/pkg/applicationserver/io"
	iogrpc "go.thethings.network/lorawan-stack/v3/pkg/applicationserver/io/grpc"
	"go.thethings.network/lorawan-stack/v3/pkg/applicationserver/io/mqtt"
	"go.thethings.network/lorawan-stack/v3/pkg/applicationserver/io/packages"
	"go.thethings.network/lorawan-stack/v3/pkg/applicationserver/io/pubsub"
	_ "go.thethings.network/lorawan-stack/v3/pkg/applicationserver/io/pubsub/provider/mqtt"  
	_ "go.thethings.network/lorawan-stack/v3/pkg/applicationserver/io/pubsub/provider/nats"  
	"go.thethings.network/lorawan-stack/v3/pkg/applicationserver/io/web"
	"go.thethings.network/lorawan-stack/v3/pkg/cluster"
	"go.thethings.network/lorawan-stack/v3/pkg/component"
	"go.thethings.network/lorawan-stack/v3/pkg/config"
	"go.thethings.network/lorawan-stack/v3/pkg/crypto/cryptoutil"
	"go.thethings.network/lorawan-stack/v3/pkg/errors"
	"go.thethings.network/lorawan-stack/v3/pkg/events"
	"go.thethings.network/lorawan-stack/v3/pkg/interop"
	"go.thethings.network/lorawan-stack/v3/pkg/log"
	"go.thethings.network/lorawan-stack/v3/pkg/messageprocessors"
	"go.thethings.network/lorawan-stack/v3/pkg/messageprocessors/cayennelpp"
	"go.thethings.network/lorawan-stack/v3/pkg/messageprocessors/devicerepository"
	"go.thethings.network/lorawan-stack/v3/pkg/messageprocessors/javascript"
	"go.thethings.network/lorawan-stack/v3/pkg/rpcmiddleware/hooks"
	"go.thethings.network/lorawan-stack/v3/pkg/ttnpb"
	"go.thethings.network/lorawan-stack/v3/pkg/types"
	"go.thethings.network/lorawan-stack/v3/pkg/unique"
	"google.golang.org/grpc"
)
type ApplicationServer struct {
	*component.Component
	ctx context.Context
	config *Config
	linkRegistry     LinkRegistry
	deviceRegistry   DeviceRegistry
	appUpsRegistry   ApplicationUplinkRegistry
	formatters       messageprocessors.MapPayloadProcessor
	webhooks         web.Webhooks
	webhookTemplates web.TemplateStore
	pubsub           *pubsub.PubSub
	appPackages      packages.Server
	clusterDistributor distribution.Distributor
	localDistributor   distribution.Distributor
	grpc struct {
		asDevices asEndDeviceRegistryServer
		appAs     ttnpb.AppAsServer
	}
	interopClient InteropClient
	interopID     string
	endDeviceFetcher EndDeviceFetcher
	activationTasks sync.Map
}
func (as *ApplicationServer) Context() context.Context {
	return as.ctx
}
var errListenFrontend = errors.DefineFailedPrecondition("listen_frontend", "failed to start frontend listener `{protocol}` on address `{address}`")
func New(c *component.Component, conf *Config) (as *ApplicationServer, err error) {
	ctx := log.NewContextWithField(c.Context(), "namespace", "applicationserver")
	baseConf := c.GetBaseConfig(ctx)
	httpClient, err := c.HTTPClient(ctx)
	if err != nil {
		return nil, err
	}
	var interopCl InteropClient
	if !conf.Interop.IsZero() {
		interopConf := conf.Interop.InteropClient
		interopConf.GetFallbackTLSConfig = func(ctx context.Context) (*tls.Config, error) {
			return c.GetTLSClientConfig(ctx)
		}
		interopConf.BlobConfig = baseConf.Blob
		if interopConf.HTTPClient == nil {
			interopConf.HTTPClient = httpClient
		}
		interopCl, err = interop.NewClient(ctx, interopConf)
		if err != nil {
			return nil, err
		}
	}
	as = &ApplicationServer{
		Component:      c,
		ctx:            ctx,
		config:         conf,
		linkRegistry:   conf.Links,
		deviceRegistry: wrapEndDeviceRegistryWithReplacedFields(conf.Devices, replacedEndDeviceFields...),
		appUpsRegistry: conf.UplinkStorage.Registry,
		formatters: messageprocessors.MapPayloadProcessor{
			ttnpb.PayloadFormatter_FORMATTER_JAVASCRIPT: javascript.New(),
			ttnpb.PayloadFormatter_FORMATTER_CAYENNELPP: cayennelpp.New(),
		},
		clusterDistributor: distribution.NewPubSubDistributor(
			ctx,
			c,
			conf.Distribution.Timeout,
			conf.Distribution.Global.PubSub,
			conf.Distribution.Global.Individual.SubscriptionOptions(),
		),
		localDistributor: distribution.NewLocalDistributor(
			ctx,
			c,
			conf.Distribution.Timeout,
			conf.Distribution.Local.Broadcast.SubscriptionOptions(),
			conf.Distribution.Local.Individual.SubscriptionOptions(),
		),
		interopClient:    interopCl,
		interopID:        conf.Interop.ID,
		endDeviceFetcher: conf.EndDeviceFetcher.Fetcher,
	}
	as.formatters[ttnpb.PayloadFormatter_FORMATTER_REPOSITORY] = devicerepository.New(as.formatters, as)
	if as.endDeviceFetcher == nil {
		as.endDeviceFetcher = &NoopEndDeviceFetcher{}
	}
	as.grpc.asDevices = asEndDeviceRegistryServer{
		AS:       as,
		kekLabel: conf.DeviceKEKLabel,
	}
	as.grpc.appAs = iogrpc.New(as,
		iogrpc.WithMQTTConfigProvider(as),
		iogrpc.WithEndDeviceFetcher(as.endDeviceFetcher),
		iogrpc.WithPayloadProcessor(as.formatters),
		iogrpc.WithSkipPayloadCrypto(func(ctx context.Context, ids ttnpb.EndDeviceIdentifiers) (bool, error) {
			link, err := as.getLink(ctx, &ids.ApplicationIdentifiers, []string{"skip_payload_crypto"})
			if err != nil {
				return false, err
			}
			dev, err := as.deviceRegistry.Get(ctx, ids, []string{"skip_payload_crypto_override"})
			if err != nil {
				return false, err
			}
			return as.skipPayloadCrypto(ctx, link, dev, nil), nil
		}),
	)
	ctx, cancel := context.WithCancel(as.Context())
	defer func() {
		if err != nil {
			cancel()
		}
	}()
	for _, version := range []struct {
		Format mqtt.Format
		Config config.MQTT
	}{
		{
			Format: mqtt.JSON,
			Config: conf.MQTT,
		},
	} {
		for _, endpoint := range []component.Endpoint{
			component.NewTCPEndpoint(version.Config.Listen, "MQTT"),
			component.NewTLSEndpoint(version.Config.ListenTLS, "MQTT"),
		} {
			version := version
			endpoint := endpoint
			if endpoint.Address() == "" {
				continue
			}
			as.RegisterTask(&component.TaskConfig{
				Context: as.Context(),
				ID:      fmt.Sprintf("serve_mqtt/%s", endpoint.Address()),
				Func: func(ctx context.Context) error {
					l, err := as.ListenTCP(endpoint.Address())
					var lis net.Listener
					if err == nil {
						lis, err = endpoint.Listen(l)
					}
					if err != nil {
						return errListenFrontend.WithCause(err).WithAttributes(
							"address", endpoint.Address(),
							"protocol", endpoint.Protocol(),
						)
					}
					defer lis.Close()
					return mqtt.Serve(ctx, as, lis, version.Format, endpoint.Protocol())
				},
				Restart: component.TaskRestartOnFailure,
				Backoff: component.DefaultTaskBackoffConfig,
			})
		}
	}
	if webhooks, err := conf.Webhooks.NewWebhooks(ctx, as); err != nil {
		return nil, err
	} else if webhooks != nil {
		as.webhooks = webhooks
		c.RegisterWeb(webhooks)
	}
	if conf.Webhooks.Templates.HTTPClient == nil {
		conf.Webhooks.Templates.HTTPClient = httpClient
	}
	if as.webhookTemplates, err = conf.Webhooks.Templates.NewTemplateStore(); err != nil {
		return nil, err
	}
	if as.pubsub, err = conf.PubSub.NewPubSub(c, as); err != nil {
		return nil, err
	}
	if as.appPackages, err = conf.Packages.NewApplicationPackages(ctx, as); err != nil {
		return nil, err
	} else if as.appPackages != nil {
		c.RegisterGRPC(as.appPackages)
	}
	hooks.RegisterUnaryHook("/ttn.lorawan.v3.NsAs", cluster.HookName, c.ClusterAuthUnaryHook())
	c.RegisterGRPC(as)
	return as, nil
}
func (as *ApplicationServer) RegisterServices(s *grpc.Server) {
	ttnpb.RegisterAsServer(s, as)
	ttnpb.RegisterNsAsServer(s, as)
	ttnpb.RegisterAsEndDeviceRegistryServer(s, as.grpc.asDevices)
	ttnpb.RegisterAppAsServer(s, as.grpc.appAs)
	if as.webhooks != nil {
		ttnpb.RegisterApplicationWebhookRegistryServer(s, web.NewWebhookRegistryRPC(as.webhooks.Registry(), as.webhookTemplates))
	}
	if as.pubsub != nil {
		ttnpb.RegisterApplicationPubSubRegistryServer(s, as.pubsub)
	}
}
func (as *ApplicationServer) RegisterHandlers(s *runtime.ServeMux, conn *grpc.ClientConn) {
	ttnpb.RegisterAsHandler(as.Context(), s, conn)
	ttnpb.RegisterAsEndDeviceRegistryHandler(as.Context(), s, conn)
	ttnpb.RegisterAppAsHandler(as.Context(), s, conn)
	if as.webhooks != nil {
		ttnpb.RegisterApplicationWebhookRegistryHandler(as.Context(), s, conn)
	}
	if as.pubsub != nil {
		ttnpb.RegisterApplicationPubSubRegistryHandler(as.Context(), s, conn)
	}
}
func (as *ApplicationServer) Roles() []ttnpb.ClusterRole {
	return []ttnpb.ClusterRole{ttnpb.ClusterRole_APPLICATION_SERVER}
}
func (as *ApplicationServer) Subscribe(ctx context.Context, protocol string, ids *ttnpb.ApplicationIdentifiers, cluster bool) (*io.Subscription, error) {
	ctx = events.ContextWithCorrelationID(ctx, fmt.Sprintf("as:conn:%s", events.NewCorrelationID()))
	if ids != nil {
		uid := unique.ID(ctx, ids)
		ctx = log.NewContextWithField(ctx, "application_uid", uid)
	}
	if cluster {
		return as.clusterDistributor.Subscribe(ctx, protocol, ids)
	}
	return as.localDistributor.Subscribe(ctx, protocol, ids)
}
func (as *ApplicationServer) Publish(ctx context.Context, up *ttnpb.ApplicationUp) error {
	link, err := as.getLink(ctx, &up.ApplicationIdentifiers, []string{
		"default_formatters",
		"skip_payload_crypto",
	})
	if err != nil {
		return err
	}
	return as.processUp(ctx, up, link)
}
func (as *ApplicationServer) processUp(ctx context.Context, up *ttnpb.ApplicationUp, link *ttnpb.ApplicationLink) error {
	ctx = log.NewContextWithField(ctx, "device_uid", unique.ID(ctx, up.EndDeviceIdentifiers))
	ctx = events.ContextWithCorrelationID(ctx, append(up.CorrelationIds, fmt.Sprintf("as:up:%s", events.NewCorrelationID()))...)
	up.CorrelationIds = events.CorrelationIDsFromContext(ctx)
	registerReceiveUp(ctx, up)
	now := time.Now().UTC()
	up.ReceivedAt = &now
	pass, err := as.handleUp(ctx, up, link)
	if err != nil {
		log.FromContext(ctx).WithError(err).Warn("Failed to process upstream message")
		registerDropUp(ctx, up, err)
		return nil
	}
	if !pass {
		return nil
	}
	if err := as.publishUp(ctx, up); err != nil {
		log.FromContext(ctx).WithError(err).Warn("Failed to broadcast upstream message")
		registerDropUp(ctx, up, err)
		return nil
	}
	registerForwardUp(ctx, up)
	return nil
}
func (as *ApplicationServer) publishUp(ctx context.Context, up *ttnpb.ApplicationUp) error {
	if err := as.localDistributor.Publish(ctx, up); err != nil {
		return err
	}
	return as.clusterDistributor.Publish(ctx, up)
}
func (as *ApplicationServer) skipPayloadCrypto(ctx context.Context, link *ttnpb.ApplicationLink, dev *ttnpb.EndDevice, session *ttnpb.Session) bool {
	if session != nil && session.AppSKey != nil {
		if _, err := cryptoutil.UnwrapAES128Key(ctx, session.AppSKey, as.KeyVault); err == nil {
			return false
		}
	}
	if dev.SkipPayloadCryptoOverride != nil {
		return dev.SkipPayloadCryptoOverride.Value
	}
	return link.SkipPayloadCrypto.GetValue()
}
func lastAFCntDownFromMinFCnt(min uint32) uint32 {
	if min == 0 {
		return 0
	}
	return min - 1
}
var (
	errDeviceNotFound  = errors.DefineNotFound("device_not_found", "device `{device_uid}` not found")
	errNoDeviceSession = errors.DefineFailedPrecondition("no_device_session", "no device session; check device activation")
	errRebuild         = errors.DefineAborted("rebuild", "could not rebuild device session; check device address")
)
func (as *ApplicationServer) buildSessionsFromError(ctx context.Context, dev *ttnpb.EndDevice, err error) ([]string, error) {
	reconstructSession := func(sessionKeyID []byte, devAddr *types.DevAddr, minFCntDown uint32) (*ttnpb.Session, error) {
		appSKey, err := as.fetchAppSKey(ctx, dev.EndDeviceIdentifiers, sessionKeyID)
		if err != nil {
			return nil, errFetchAppSKey.WithCause(err)
		}
		return &ttnpb.Session{
			DevAddr: *devAddr,
			SessionKeys: ttnpb.SessionKeys{
				SessionKeyId: sessionKeyID,
				AppSKey:      &appSKey,
			},
			LastAFCntDown: lastAFCntDownFromMinFCnt(minFCntDown),
			StartedAt:     time.Now().UTC(),
		}, nil
	}
	ttnErr, ok := err.(errors.ErrorDetails)
	if !ok {
		return nil, err
	}
	details := ttnErr.Details()
	if len(details) == 0 {
		return nil, err
	}
	var diagnostics *ttnpb.DownlinkQueueOperationErrorDetails
	for _, detail := range details {
		diagnostics, ok = detail.(*ttnpb.DownlinkQueueOperationErrorDetails)
		if ok {
			break
		}
	}
	if diagnostics == nil {
		return nil, err
	}
	var mask []string
	if diagnostics.DevAddr != nil {
		switch {
		case dev.Session != nil &&
			bytes.Equal(diagnostics.SessionKeyId, dev.Session.SessionKeyId) &&
			dev.Session.DevAddr.Equal(*diagnostics.DevAddr):
			dev.Session.LastAFCntDown = lastAFCntDownFromMinFCnt(diagnostics.MinFCntDown)
		case len(diagnostics.SessionKeyId) > 0:
			session, err := reconstructSession(diagnostics.SessionKeyId, diagnostics.DevAddr, diagnostics.MinFCntDown)
			if err != nil {
				return nil, err
			}
			dev.Session = session
			dev.DevAddr = &session.DevAddr
		default:
			return nil, errRebuild.New()
		}
	} else {
		dev.Session = nil
		dev.DevAddr = nil
	}
	mask = ttnpb.AddFields(mask, "session", "ids.dev_addr")
	if diagnostics.PendingDevAddr != nil {
		switch {
		case dev.PendingSession != nil &&
			bytes.Equal(diagnostics.PendingSessionKeyId, dev.PendingSession.SessionKeyId) &&
			dev.PendingSession.DevAddr.Equal(*diagnostics.PendingDevAddr):
			dev.PendingSession.LastAFCntDown = lastAFCntDownFromMinFCnt(diagnostics.PendingMinFCntDown)
		case len(diagnostics.PendingSessionKeyId) > 0:
			session, err := reconstructSession(diagnostics.PendingSessionKeyId, diagnostics.PendingDevAddr, diagnostics.PendingMinFCntDown)
			if err != nil {
				return nil, err
			}
			dev.PendingSession = session
		default:
			return nil, errRebuild.New()
		}
	} else {
		dev.PendingSession = nil
	}
	mask = ttnpb.AddFields(mask, "pending_session")
	return mask, nil
}
type downlinkQueueOperation struct {
	Items             []*ttnpb.ApplicationDownlink
	Operation         func(ttnpb.AsNsClient, context.Context, *ttnpb.DownlinkQueueRequest, ...grpc.CallOption) (*pbtypes.Empty, error)
	SkipSessionKeyIDs [][]byte
}
func (d downlinkQueueOperation) shouldSkip(sessionKeyID []byte) bool {
	for _, id := range d.SkipSessionKeyIDs {
		if bytes.Equal(id, sessionKeyID) {
			return true
		}
	}
	return false
}
const maxDownlinkQueueOperationAttempts = 50
func (as *ApplicationServer) attemptDownlinkQueueOp(ctx context.Context, dev *ttnpb.EndDevice, link *ttnpb.ApplicationLink, peer cluster.Peer, op downlinkQueueOperation) ([]string, error) {
	pc, err := peer.Conn()
	if err != nil {
		return nil, err
	}
	mask := make([]string, 0, 2)
	attempt := 1
	for {
		ctx := log.NewContextWithField(ctx, "attempt", attempt)
		sessions := make([]*ttnpb.Session, 0, 2)
		if dev.Session != nil && !op.shouldSkip(dev.Session.SessionKeyId) {
			sessions = append(sessions, dev.Session)
			mask = ttnpb.AddFields(mask, "session.last_a_f_cnt_down")
		}
		if dev.PendingSession != nil && !op.shouldSkip(dev.PendingSession.SessionKeyId) {
			sessions = append(sessions, dev.PendingSession)
			mask = ttnpb.AddFields(mask, "pending_session.last_a_f_cnt_down")
		}
		if len(sessions) == 0 {
			return nil, errNoDeviceSession.New()
		}
		encryptedItems, err := as.encodeAndEncryptDownlinks(ctx, dev, link, op.Items, sessions)
		if err != nil {
			return nil, err
		}
		_, err = op.Operation(ttnpb.NewAsNsClient(pc), ctx, &ttnpb.DownlinkQueueRequest{
			EndDeviceIdentifiers: dev.EndDeviceIdentifiers,
			Downlinks:            encryptedItems,
		}, as.WithClusterAuth())
		if err == nil {
			return mask, nil
		}
		if attempt >= maxDownlinkQueueOperationAttempts || as.skipPayloadCrypto(ctx, link, dev, nil) {
			return nil, err
		}
		mask, err = as.buildSessionsFromError(ctx, dev, err)
		if err != nil {
			return nil, err
		}
		attempt++
	}
}
func (as *ApplicationServer) downlinkQueueOp(ctx context.Context, ids ttnpb.EndDeviceIdentifiers, items []*ttnpb.ApplicationDownlink, op func(ttnpb.AsNsClient, context.Context, *ttnpb.DownlinkQueueRequest, ...grpc.CallOption) (*pbtypes.Empty, error)) error {
	ctx = events.ContextWithCorrelationID(ctx, fmt.Sprintf("as:downlink:%s", events.NewCorrelationID()))
	link, err := as.getLink(ctx, &ids.ApplicationIdentifiers, []string{
		"default_formatters",
		"skip_payload_crypto",
	})
	if err != nil {
		return err
	}
	peer, err := as.GetPeer(ctx, ttnpb.ClusterRole_NETWORK_SERVER, &ids)
	if err != nil {
		return err
	}
	for _, item := range items {
		item.CorrelationIds = append(item.CorrelationIds, events.CorrelationIDsFromContext(ctx)...)
	}
	registerReceiveDownlinks(ctx, ids, items)
	_, err = as.deviceRegistry.Set(ctx, ids,
		[]string{
			"formatters",
			"pending_session",
			"session",
			"skip_payload_crypto_override",
			"version_ids",
		},
		func(dev *ttnpb.EndDevice) (*ttnpb.EndDevice, []string, error) {
			if dev == nil {
				return nil, nil, errDeviceNotFound.WithAttributes("device_uid", unique.ID(ctx, ids))
			}
			mask, err := as.attemptDownlinkQueueOp(ctx, dev, link, peer, downlinkQueueOperation{
				Items:     items,
				Operation: op,
			})
			if err != nil {
				return nil, nil, err
			}
			return dev, mask, nil
		},
	)
	if err != nil {
		as.registerDropDownlinks(ctx, ids, items, err)
		return err
	}
	as.registerForwardDownlinks(ctx, ids, items, peer.Name())
	return nil
}
func (as *ApplicationServer) DownlinkQueuePush(ctx context.Context, ids ttnpb.EndDeviceIdentifiers, items []*ttnpb.ApplicationDownlink) error {
	return as.downlinkQueueOp(ctx, ids, io.CleanDownlinks(items), ttnpb.AsNsClient.DownlinkQueuePush)
}
func (as *ApplicationServer) DownlinkQueueReplace(ctx context.Context, ids ttnpb.EndDeviceIdentifiers, items []*ttnpb.ApplicationDownlink) error {
	return as.downlinkQueueOp(ctx, ids, io.CleanDownlinks(items), ttnpb.AsNsClient.DownlinkQueueReplace)
}
var errNoAppSKey = errors.DefineCorruption("no_app_s_key", "no AppSKey")
func (as *ApplicationServer) DownlinkQueueList(ctx context.Context, ids ttnpb.EndDeviceIdentifiers) ([]*ttnpb.ApplicationDownlink, error) {
	dev, err := as.deviceRegistry.Get(ctx, ids, []string{
		"formatters",
		"session",
		"skip_payload_crypto",
		"pending_session",
	})
	if err != nil {
		return nil, err
	}
	link, err := as.getLink(ctx, &ids.ApplicationIdentifiers, []string{
		"default_formatters",
		"skip_payload_crypto",
	})
	if err != nil {
		return nil, err
	}
	pc, err := as.GetPeerConn(ctx, ttnpb.ClusterRole_NETWORK_SERVER, &ids)
	if err != nil {
		return nil, err
	}
	client := ttnpb.NewAsNsClient(pc)
	res, err := client.DownlinkQueueList(ctx, &ids, as.WithClusterAuth())
	if err != nil {
		return nil, err
	}
	if len(res.Downlinks) == 0 {
		return nil, nil
	}
	var queue []*ttnpb.ApplicationDownlink
	var session *ttnpb.Session
	switch {
	case dev.Session != nil:
		session = dev.Session
	case dev.PendingSession != nil:
		session = dev.PendingSession
	default:
		return nil, errNoDeviceSession.New()
	}
	if session.AppSKey == nil {
		return nil, errNoAppSKey.New()
	}
	queue, _ = ttnpb.PartitionDownlinksBySessionKeyIDEquality(session.SessionKeyId, res.Downlinks...)
	if as.skipPayloadCrypto(ctx, link, dev, session) {
		return queue, nil
	}
	for _, item := range queue {
		if err := as.decryptAndDecodeDownlink(ctx, dev, item, link.DefaultFormatters); err != nil {
			return nil, err
		}
	}
	return queue, nil
}
var (
	errJSUnavailable = errors.DefineUnavailable("join_server_unavailable", "Join Server unavailable for JoinEUI `{join_eui}`")
	errNoDevEUI      = errors.DefineInvalidArgument("no_dev_eui", "no device EUI provided")
	errNoJoinEUI     = errors.DefineInvalidArgument("no_join_eui", "no join EUI provided")
)
func (as *ApplicationServer) fetchAppSKey(ctx context.Context, ids ttnpb.EndDeviceIdentifiers, sessionKeyID []byte) (ttnpb.KeyEnvelope, error) {
	if ids.DevEui == nil {
		return ttnpb.KeyEnvelope{}, errNoDevEUI.New()
	}
	if ids.JoinEui == nil {
		return ttnpb.KeyEnvelope{}, errNoJoinEUI.New()
	}
	req := &ttnpb.SessionKeyRequest{
		SessionKeyId: sessionKeyID,
		DevEui:       *ids.DevEui,
		JoinEui:      *ids.JoinEui,
	}
	if js, err := as.GetPeer(ctx, ttnpb.ClusterRole_JOIN_SERVER, &ids); err == nil {
		cc, err := js.Conn()
		if err != nil {
			return ttnpb.KeyEnvelope{}, err
		}
		res, err := ttnpb.NewAsJsClient(cc).GetAppSKey(ctx, req, as.WithClusterAuth())
		if err == nil && res.AppSKey != nil {
			return *res.AppSKey, nil
		}
		if !errors.IsNotFound(err) {
			return ttnpb.KeyEnvelope{}, err
		}
	}
	if as.interopClient != nil && !interop.GeneratedSessionKeyID(sessionKeyID) {
		res, err := as.interopClient.GetAppSKey(ctx, as.interopID, req)
		if err == nil && res.AppSKey != nil {
			return *res.AppSKey, nil
		}
		if !errors.IsNotFound(err) {
			return ttnpb.KeyEnvelope{}, err
		}
	}
	return ttnpb.KeyEnvelope{}, errJSUnavailable.WithAttributes("join_eui", *ids.JoinEui)
}
func (as *ApplicationServer) handleUp(ctx context.Context, up *ttnpb.ApplicationUp, link *ttnpb.ApplicationLink) (pass bool, err error) {
	if up.Simulated {
		return true, as.handleSimulatedUp(ctx, up, link)
	}
	switch p := up.Up.(type) {
	case *ttnpb.ApplicationUp_JoinAccept:
		return true, as.handleJoinAccept(ctx, up.EndDeviceIdentifiers, p.JoinAccept, link)
	case *ttnpb.ApplicationUp_UplinkMessage:
		return true, as.handleUplink(ctx, up.EndDeviceIdentifiers, p.UplinkMessage, link)
	case *ttnpb.ApplicationUp_DownlinkQueueInvalidated:
		return as.handleDownlinkQueueInvalidated(ctx, up.EndDeviceIdentifiers, p.DownlinkQueueInvalidated, link)
	case *ttnpb.ApplicationUp_DownlinkQueued:
		return true, as.decryptDownlinkMessage(ctx, up.EndDeviceIdentifiers, p.DownlinkQueued, link)
	case *ttnpb.ApplicationUp_DownlinkSent:
		return true, as.decryptDownlinkMessage(ctx, up.EndDeviceIdentifiers, p.DownlinkSent, link)
	case *ttnpb.ApplicationUp_DownlinkFailed:
		return true, as.decryptDownlinkMessage(ctx, up.EndDeviceIdentifiers, &p.DownlinkFailed.ApplicationDownlink, link)
	case *ttnpb.ApplicationUp_DownlinkAck:
		return true, as.decryptDownlinkMessage(ctx, up.EndDeviceIdentifiers, p.DownlinkAck, link)
	case *ttnpb.ApplicationUp_DownlinkNack:
		return true, as.handleDownlinkNack(ctx, up.EndDeviceIdentifiers, p.DownlinkNack, link)
	case *ttnpb.ApplicationUp_LocationSolved:
		return true, as.handleLocationSolved(ctx, up.EndDeviceIdentifiers, p.LocationSolved, link)
	case *ttnpb.ApplicationUp_ServiceData:
		return true, nil
	default:
		return false, nil
	}
}
func (as *ApplicationServer) handleSimulatedUp(ctx context.Context, up *ttnpb.ApplicationUp, link *ttnpb.ApplicationLink) error {
	switch p := up.Up.(type) {
	case *ttnpb.ApplicationUp_UplinkMessage:
		return as.handleSimulatedUplink(ctx, up.EndDeviceIdentifiers, p.UplinkMessage, link)
	default:
		return nil
	}
}
var errFetchAppSKey = errors.Define("app_s_key", "failed to get AppSKey")
func (as *ApplicationServer) handleJoinAccept(ctx context.Context, ids ttnpb.EndDeviceIdentifiers, joinAccept *ttnpb.ApplicationJoinAccept, link *ttnpb.ApplicationLink) error {
	logger := log.FromContext(ctx).WithFields(log.Fields(
		"join_eui", ids.JoinEui,
		"dev_eui", ids.DevEui,
		"session_key_id", joinAccept.SessionKeyId,
	))
	peer, err := as.GetPeer(ctx, ttnpb.ClusterRole_NETWORK_SERVER, &ids)
	if err != nil {
		return err
	}
	_, err = as.deviceRegistry.Set(ctx, ids,
		[]string{
			"formatters",
			"pending_session",
			"session",
			"skip_payload_crypto_override",
			"version_ids",
		},
		func(dev *ttnpb.EndDevice) (*ttnpb.EndDevice, []string, error) {
			var mask []string
			if dev == nil {
				return nil, nil, errDeviceNotFound.WithAttributes("device_uid", unique.ID(ctx, ids))
			}
			if joinAccept.AppSKey != nil {
				logger.Debug("Received AppSKey from Network Server")
			} else {
				logger.Debug("Fetch AppSKey from Join Server")
				key, err := as.fetchAppSKey(ctx, ids, joinAccept.SessionKeyId)
				if err != nil {
					return nil, nil, errFetchAppSKey.WithCause(err)
				}
				joinAccept.AppSKey = &key
				logger.Debug("Fetched AppSKey from Join Server")
			}
			previousSession := dev.PendingSession
			dev.PendingSession = &ttnpb.Session{
				DevAddr: *ids.DevAddr,
				SessionKeys: ttnpb.SessionKeys{
					SessionKeyId: joinAccept.SessionKeyId,
					AppSKey:      joinAccept.AppSKey,
				},
				StartedAt: time.Now().UTC(),
			}
			mask = ttnpb.AddFields(mask, "pending_session")
			if as.skipPayloadCrypto(ctx, link, dev, dev.PendingSession) {
				return dev, mask, nil
			}
			joinAccept.AppSKey = nil
			if len(joinAccept.InvalidatedDownlinks) == 0 {
				return dev, mask, nil
			}
			logger := logger.WithField("count", len(joinAccept.InvalidatedDownlinks))
			logger.Debug("Recalculating downlink queue to restore downlink queue on join")
			items := make([]*ttnpb.ApplicationDownlink, 0, len(joinAccept.InvalidatedDownlinks))
			for _, msg := range joinAccept.InvalidatedDownlinks {
				if err := as.decryptDownlink(ctx, dev, msg, previousSession); err != nil {
					logger.WithError(err).Warn("Failed to decrypt downlink message; drop item")
					registerDropDownlink(ctx, ids, msg, err)
					continue
				}
				items = append(items, msg)
			}
			joinAccept.InvalidatedDownlinks = nil
			if len(items) == 0 {
				return dev, mask, nil
			}
			pushMask, err := as.attemptDownlinkQueueOp(ctx, dev, link, peer, downlinkQueueOperation{
				Items:     items,
				Operation: ttnpb.AsNsClient.DownlinkQueuePush,
				SkipSessionKeyIDs: [][]byte{items[0].SessionKeyId},
			})
			if err != nil {
				as.registerDropDownlinks(ctx, ids, items, err)
				return nil, nil, err
			}
			mask = ttnpb.AddFields(mask, pushMask...)
			return dev, mask, nil
		},
	)
	return err
}
var errUnknownSession = errors.DefineNotFound("unknown_session", "unknown session")