grafana/pkg/registry/apis/dashboard/legacy/storage.go

package legacy

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"iter"
	"net/http"
	"time"

	claims "github.com/grafana/authlib/types"

	dashboard "github.com/grafana/grafana/apps/dashboard/pkg/apis/dashboard/v1beta1"
	"github.com/grafana/grafana/pkg/apimachinery/utils"
	"github.com/grafana/grafana/pkg/services/dashboards"
	"github.com/grafana/grafana/pkg/storage/unified/resource"
	"github.com/grafana/grafana/pkg/storage/unified/resourcepb"

	apierrors "k8s.io/apimachinery/pkg/api/errors"
)

func getDashboardFromEvent(event resource.WriteEvent) (*dashboard.Dashboard, error) {
	obj, ok := event.Object.GetRuntimeObject()
	if ok && obj != nil {
		dash, ok := obj.(*dashboard.Dashboard)
		if ok {
			return dash, nil
		}
	}
	dash := &dashboard.Dashboard{}
	err := json.Unmarshal(event.Value, dash)
	return dash, err
}

func getProvisioningDataFromEvent(event resource.WriteEvent) (*dashboards.DashboardProvisioning, error) {
	obj, ok := event.Object.GetRuntimeObject()
	if !ok {
		return nil, fmt.Errorf("object is not a runtime object")
	}
	meta, err := utils.MetaAccessor(obj)
	if err != nil {
		return nil, err
	}

	provisioningData, ok := meta.GetManagerProperties()
	if !ok || (provisioningData.Kind != utils.ManagerKindClassicFP) { //nolint:staticcheck
		return nil, nil
	}
	source, ok := meta.GetSourceProperties()
	if !ok {
		return nil, nil
	}
	provisioning := &dashboards.DashboardProvisioning{
		Name:       provisioningData.Identity,
		ExternalID: source.Path,
		CheckSum:   source.Checksum,
	}
	if source.TimestampMillis > 0 {
		provisioning.Updated = time.UnixMilli(source.TimestampMillis).Unix()
	}

	return provisioning, nil
}

func isDashboardKey(key *resourcepb.ResourceKey, requireName bool) error {
	gr := dashboard.DashboardResourceInfo.GroupResource()
	if key.Group != gr.Group {
		return fmt.Errorf("expecting dashboard group (%s != %s)", key.Group, gr.Group)
	}
	if key.Resource != gr.Resource {
		return fmt.Errorf("expecting dashboard resource (%s != %s)", key.Resource, gr.Resource)
	}
	if requireName && key.Name == "" {
		return fmt.Errorf("expecting dashboard name (uid)")
	}
	return nil
}

func (a *dashboardSqlAccess) WriteEvent(ctx context.Context, event resource.WriteEvent) (rv int64, err error) {
	ctx, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.WriteEvent")
	defer span.End()

	info, err := claims.ParseNamespace(event.Key.Namespace)
	if err == nil {
		err = isDashboardKey(event.Key, true)
	}
	if err != nil {
		return 0, err
	}

	switch event.Type {
	case resourcepb.WatchEvent_DELETED:
		{
			_, _, err = a.DeleteDashboard(ctx, info.OrgID, event.Key.Name)
			//rv = ???
		}
	// The difference depends on embedded internal ID
	case resourcepb.WatchEvent_ADDED, resourcepb.WatchEvent_MODIFIED:
		{
			dash, err := getDashboardFromEvent(event)
			if err != nil {
				return 0, err
			}
			// In unistore, provisioning data is stored as annotations on the dashboard object. In legacy, it is stored in a separate
			// database table. For the legacy fallback, we need to save the provisioning data in the same transaction - so we need to handle these separately.
			// Without this, we can end up having dashboards created in legacy, unistore timing out, and then never saving the provisioning data, which
			// results in duplicated dashboards on next startup.
			provisioning, err := getProvisioningDataFromEvent(event)
			if err != nil {
				return 0, err
			}
			if provisioning != nil {
				cmd, _, err := a.buildSaveDashboardCommand(ctx, info.OrgID, dash)
				if err != nil {
					return 0, err
				}

				after, err := a.dashStore.SaveProvisionedDashboard(ctx, *cmd, provisioning)
				if err != nil {
					return 0, err
				}

				// dashboard version is the RV in legacy storage
				// and deprecatedInternalID must be set here (as SaveDashboard does below for non-provisioned dashboards)
				if after != nil {
					rv = int64(after.Version)
					access := GetLegacyAccess(ctx)
					if access != nil {
						access.DashboardID = after.ID
					}
				}
			} else {
				failOnExisting := event.Type == resourcepb.WatchEvent_ADDED
				sql, err := a.sql(ctx)
				if err != nil {
					return 0, err
				}

				var after *dashboard.Dashboard
				if err := sql.DB.InTransaction(ctx, func(ctx context.Context) error {
					var err error
					after, _, err = a.SaveDashboard(ctx, info.OrgID, dash, failOnExisting)
					return err
				}); err != nil {
					return 0, err
				}
				if after != nil {
					meta, err := utils.MetaAccessor(after)
					if err != nil {
						return 0, err
					}
					rv, err = meta.GetResourceVersionInt64()
					if err != nil {
						return 0, err
					}
				}
			}
		}
	default:
		return 0, fmt.Errorf("unsupported event type: %v", event.Type)
	}

	// Async notify all subscribers (not HA!!!)
	if a.subscribers != nil {
		go func() {
			write := &resource.WrittenEvent{
				Type:       event.Type,
				Key:        event.Key,
				PreviousRV: event.PreviousRV,
				Value:      event.Value,

				Timestamp:       time.Now().UnixMilli(),
				ResourceVersion: rv,
			}
			for _, sub := range a.subscribers {
				sub <- write
			}
		}()
	}
	return rv, err
}

func (a *dashboardSqlAccess) GetDashboard(ctx context.Context, orgId int64, uid string, v int64) (*dashboard.Dashboard, int64, error) {
	ctx, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.GetDashboard")
	defer span.End()

	sql, err := a.sql(ctx)
	if err != nil {
		return nil, 0, err
	}

	rows, err := a.getRows(ctx, sql, &DashboardQuery{
		OrgID:   orgId,
		UID:     uid,
		Limit:   2, // will only be one!
		Version: v,
	})
	if err != nil {
		return nil, 0, err
	}
	defer func() { _ = rows.Close() }()

	if rows.Next() {
		row := rows.row
		if row != nil {
			return row.Dash, row.RV, rows.err
		}
	}
	return nil, 0, rows.err
}

// Read implements ResourceStoreServer.
func (a *dashboardSqlAccess) ReadResource(ctx context.Context, req *resourcepb.ReadRequest) *resource.BackendReadResponse {
	ctx, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.ReadResource")
	defer span.End()

	rsp := &resource.BackendReadResponse{}
	info, err := claims.ParseNamespace(req.Key.Namespace)
	if err == nil {
		err = isDashboardKey(req.Key, true)
	}
	if err != nil {
		rsp.Error = resource.AsErrorResult(err)
		return rsp
	}
	version := int64(0)
	if req.ResourceVersion > 0 {
		version = req.ResourceVersion
	}

	dash, rv, err := a.GetDashboard(ctx, info.OrgID, req.Key.Name, version)
	if err != nil {
		rsp.Error = resource.AsErrorResult(err)
		return rsp
	}
	if dash == nil {
		rsp.Error = &resourcepb.ErrorResult{
			Code: http.StatusNotFound,
		}
	} else {
		meta, err := utils.MetaAccessor(dash)
		if err != nil {
			rsp.Error = resource.AsErrorResult(err)
		}
		rsp.Folder = meta.GetFolder()

		rsp.Value, err = json.Marshal(dash)
		if err != nil {
			rsp.Error = resource.AsErrorResult(err)
		}
	}
	rsp.ResourceVersion = rv
	return rsp
}

// ListHistory implements StorageBackend.
func (a *dashboardSqlAccess) ListHistory(ctx context.Context, req *resourcepb.ListRequest, cb func(resource.ListIterator) error) (int64, error) {
	ctx, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.ListHistory")
	defer span.End()

	return a.ListIterator(ctx, req, cb)
}

func (a *dashboardSqlAccess) ListModifiedSince(ctx context.Context, key resource.NamespacedResource, sinceRv int64) (int64, iter.Seq2[*resource.ModifiedResource, error]) {
	_, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.ListModifiedSince")
	defer span.End()

	return 0, func(yield func(*resource.ModifiedResource, error) bool) {
		yield(nil, errors.New("not implemented"))
	}
}

func (a *dashboardSqlAccess) GetResourceLastImportTimes(ctx context.Context) iter.Seq2[resource.ResourceLastImportTime, error] {
	_, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.GetResourceLastImportTimes")
	defer span.End()

	return func(yield func(resource.ResourceLastImportTime, error) bool) {
		yield(resource.ResourceLastImportTime{}, errors.New("not implemented"))
	}
}

// List implements StorageBackend.
func (a *dashboardSqlAccess) ListIterator(ctx context.Context, req *resourcepb.ListRequest, cb func(resource.ListIterator) error) (int64, error) {
	ctx, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.ListIterator")
	defer span.End()

	if req.ResourceVersion != 0 {
		return 0, apierrors.NewBadRequest("List with explicit resourceVersion is not supported with this storage backend")
	}
	opts := req.Options
	info, err := claims.ParseNamespace(opts.Key.Namespace)
	if err == nil {
		err = isDashboardKey(opts.Key, false)
	}
	if err != nil {
		return 0, err
	}

	token, err := readContinueToken(req.NextPageToken)
	if err != nil {
		return 0, err
	}
	if token.orgId > 0 && token.orgId != info.OrgID {
		return 0, fmt.Errorf("token and orgID mismatch")
	}

	// Default batch size for iterator - fetch rows in batches to avoid slow queries
	const defaultMaxRows = 500
	maxRows := defaultMaxRows
	if req.Limit > 0 && req.Limit < int64(defaultMaxRows) {
		maxRows = int(req.Limit)
	}

	query := &DashboardQuery{
		OrgID:   info.OrgID,
		Limit:   int(req.Limit),
		MaxRows: maxRows,
		LastID:  token.id,
		Labels:  req.Options.Labels,
	}

	sql, err := a.sql(ctx)
	if err != nil {
		return 0, err
	}

	switch req.Source {
	case resourcepb.ListRequest_HISTORY:
		query.GetHistory = true
		query.UID = req.Options.Key.Name
	case resourcepb.ListRequest_TRASH:
		query.GetTrash = true
	case resourcepb.ListRequest_STORE:
		// normal
	}

	listRV, err := sql.GetResourceVersion(ctx, "dashboard", "updated")
	if err != nil {
		return 0, err
	}
	listRV *= 1000 // Convert to microseconds

	iter, err := newBatchingIterator(ctx, a, sql, query)
	if iter != nil {
		defer func() {
			_ = iter.Close()
		}()
	}
	if err == nil {
		err = cb(iter)
	}
	return listRV, err
}

// Watch implements AppendingStore.
func (a *dashboardSqlAccess) WatchWriteEvents(ctx context.Context) (<-chan *resource.WrittenEvent, error) {
	stream := make(chan *resource.WrittenEvent, 10)
	{
		a.mutex.Lock()
		defer a.mutex.Unlock()

		// Add the event stream
		a.subscribers = append(a.subscribers, stream)
	}

	// Wait for context done
	go func() {
		// Wait till the context is done
		<-ctx.Done()

		// Then remove the subscription
		a.mutex.Lock()
		defer a.mutex.Unlock()

		// Copy all streams without our listener
		subs := []chan *resource.WrittenEvent{}
		for _, sub := range a.subscribers {
			if sub != stream {
				subs = append(subs, sub)
			}
		}
		a.subscribers = subs
	}()
	return stream, nil
}

// Simple wrapper for index implementation
func (a *dashboardSqlAccess) Read(ctx context.Context, req *resourcepb.ReadRequest) (*resource.BackendReadResponse, error) {
	ctx, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.Read")
	defer span.End()

	return a.ReadResource(ctx, req), nil
}

func (a *dashboardSqlAccess) Search(ctx context.Context, req *resourcepb.ResourceSearchRequest) (*resourcepb.ResourceSearchResponse, error) {
	ctx, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.Search")
	defer span.End()

	return a.dashboardSearchClient.Search(ctx, req)
}

func (a *dashboardSqlAccess) ListManagedObjects(ctx context.Context, req *resourcepb.ListManagedObjectsRequest) (*resourcepb.ListManagedObjectsResponse, error) {
	return nil, fmt.Errorf("not implemented")
}

func (a *dashboardSqlAccess) CountManagedObjects(context.Context, *resourcepb.CountManagedObjectsRequest) (*resourcepb.CountManagedObjectsResponse, error) {
	return nil, fmt.Errorf("not implemented")
}

// GetStats implements ResourceServer.
func (a *dashboardSqlAccess) GetStats(ctx context.Context, req *resourcepb.ResourceStatsRequest) (*resourcepb.ResourceStatsResponse, error) {
	ctx, span := tracer.Start(ctx, "legacy.dashboardSqlAccess.GetStats")
	defer span.End()

	return a.dashboardSearchClient.GetStats(ctx, req)
}