Permify / permify

import (

	"context"

	"errors"

	"sync"

	"github.com/Permify/permify/internal/invoke"

	"github.com/Permify/permify/internal/schema"

	"github.com/Permify/permify/internal/storage"

	base "github.com/Permify/permify/pkg/pb/base/v1"

)

type LookupEngine struct {

	// schemaReader is responsible for reading schema information

	schemaReader storage.SchemaReader

	// schemaReader is responsible for reading data

	dataReader storage.DataReader

	// checkEngine is responsible for performing permission checks

	checkEngine invoke.Check

	// schemaMap is a map that keeps track of schema versions

	schemaMap sync.Map

	// concurrencyLimit is the maximum number of concurrent permission checks allowed

	concurrencyLimit int

}

func NewLookupEngine(

	check invoke.Check,

	schemaReader storage.SchemaReader,

	dataReader storage.DataReader,

	opts ...LookupOption,

) *LookupEngine {

	engine := &LookupEngine{

		schemaReader:     schemaReader,

		checkEngine:      check,

		dataReader:       dataReader,

		schemaMap:        sync.Map{},

		concurrencyLimit: _defaultConcurrencyLimit,

	}

	// options

	for _, opt := range opts {

		opt(engine)

	}

	return engine

}

// LookupEntity performs a permission check on a set of entities and returns a response

// containing the IDs of the entities that have the requested permission.

func (engine *LookupEngine) LookupEntity(ctx context.Context, request *base.PermissionLookupEntityRequest) (response *base.PermissionLookupEntityResponse, err error) {

	// A mutex and slice are declared to safely store entity IDs from concurrent callbacks

	var mu sync.Mutex

	var entityIDs []string

	var ct string

	size := request.GetPageSize()

	if size == 0 {

		size = 1000

	}

	// Callback function which is called for each entity. If the entity passes the permission check,

	// the entity ID is appended to the entityIDs slice.

	callback := func(entityID, token string) {

		mu.Lock()         // Safeguard access to the shared slice with a mutex

		defer mu.Unlock() // Ensure the lock is released after appending the ID

		entityIDs = append(entityIDs, entityID)

		ct = token

	}

	// Create and start BulkChecker. It performs permission checks in parallel.

	checker := NewBulkChecker(ctx, engine.checkEngine, BULK_ENTITY, callback, engine.concurrencyLimit)

	// Create and start BulkPublisher. It receives entities and passes them to BulkChecker.

	publisher := NewBulkEntityPublisher(ctx, request, checker)

	// Retrieve the schema of the entity based on the tenantId and schema version

	var sc *base.SchemaDefinition

	sc, err = engine.readSchema(ctx, request.GetTenantId(), request.GetMetadata().GetSchemaVersion())

	if err != nil {

		return nil, err

	}

	// Create a map to keep track of visited entities

	visits := &VisitsMap{}

	// Perform an entity filter operation based on the permission request

	err = NewEntityFilter(engine.dataReader, sc).EntityFilter(ctx, &base.PermissionEntityFilterRequest{

		TenantId: request.GetTenantId(),

		Metadata: &base.PermissionEntityFilterRequestMetadata{

			SnapToken:     request.GetMetadata().GetSnapToken(),

			SchemaVersion: request.GetMetadata().GetSchemaVersion(),

			Depth:         request.GetMetadata().GetDepth(),

		},

		Entrance: &base.Entrance{

			Type:  request.GetEntityType(),

			Value: request.GetPermission(),

		},

		Subject: request.GetSubject(),

		Context: request.GetContext(),

		Scope:   request.GetScope(),

		Cursor:  request.GetContinuousToken(),

	}, visits, publisher)

	if err != nil {

		return nil, err

	}

	// At this point, the BulkChecker has collected and sorted requests

	err = checker.ExecuteRequests(size) // Execute the collected requests in parallel

	if err != nil {

		return nil, err

	}

	// Return response containing allowed entity IDs

	return &base.PermissionLookupEntityResponse{

		EntityIds:       entityIDs,

		ContinuousToken: ct,

	}, nil

}

// LookupEntityStream performs a permission check on a set of entities and streams the results

// containing the IDs of the entities that have the requested permission.

// LookupSubject checks if a subject has a particular permission based on the schema and version.

// It returns a list of subjects that have the given permission.

func (engine *LookupEngine) LookupSubject(ctx context.Context, request *base.PermissionLookupSubjectRequest) (response *base.PermissionLookupSubjectResponse, err error) {

	size := request.GetPageSize()

	if size == 0 {

		size = 1000

	}

	// Use a mutex to protect concurrent writes to the subjectIDs slice.

	var mu sync.Mutex

	var subjectIDs []string

	var ct string

	// Callback function to handle the results of permission checks.

	// If an entity passes the permission check, its ID is stored in the subjectIDs slice.

	callback := func(subjectID, token string) {

		mu.Lock()         // Lock to prevent concurrent modification of the slice.

		defer mu.Unlock() // Unlock after the ID is appended.

		subjectIDs = append(subjectIDs, subjectID)

		ct = token

	}

	// Create and initiate a BulkChecker to perform permission checks in parallel.

	checker := NewBulkChecker(ctx, engine.checkEngine, BULK_SUBJECT, callback, engine.concurrencyLimit)

	// Create and start a BulkPublisher to provide entities to the BulkChecker.

	publisher := NewBulkSubjectPublisher(ctx, request, checker)

	// Retrieve the schema of the entity based on the provided tenantId and schema version.

	var sc *base.SchemaDefinition

	sc, err = engine.readSchema(ctx, request.GetTenantId(), request.GetMetadata().GetSchemaVersion())

	if err != nil {

		// Return an error if there was an issue retrieving the schema.

		return nil, err

	}

	// Walk the entity schema to perform a permission check.

	err = schema.NewWalker(sc).Walk(request.GetEntity().GetType(), request.GetPermission())

	if err != nil {

		// If the error indicates the schema walk is unimplemented, handle it with a MassEntityFilter.

		if errors.Is(err, schema.ErrUnimplemented) {

			err = NewMassSubjectFilter(engine.dataReader).SubjectFilter(ctx, request, publisher)

			if err != nil {

				// Return an error if there was an issue with the subject filter.

				return nil, err

			}

		} else { // For other errors, simply return the error

			return nil, err

		}

	} else {

		// Use the schema-based subject filter to get the list of subjects with the requested permission.

		ids, err := NewSchemaBasedSubjectFilter(engine.schemaReader, engine.dataReader, SchemaBaseSubjectFilterConcurrencyLimit(engine.concurrencyLimit)).SubjectFilter(ctx, request)

		if err != nil {

			return nil, err

		}

		for _, id := range ids {

			publisher.Publish(&base.Subject{

				Type:     request.GetSubjectReference().GetType(),

				Id:       id,

				Relation: request.GetSubjectReference().GetRelation(),

			}, &base.PermissionCheckRequestMetadata{

				SnapToken:     request.GetMetadata().GetSnapToken(),

				SchemaVersion: request.GetMetadata().GetSchemaVersion(),

				Depth:         request.GetMetadata().GetDepth(),

			}, request.GetContext(), base.CheckResult_CHECK_RESULT_ALLOWED)

		}

	}

	err = checker.ExecuteRequests(size)

	if err != nil {

		// Return an error if there was an issue with the subject filter.

		return nil, err

	}

	// Return the list of entity IDs that have the required permission.

	return &base.PermissionLookupSubjectResponse{

		SubjectIds:      subjectIDs,

		ContinuousToken: ct,

	}, nil

}

// readSchema retrieves a SchemaDefinition for a given tenantID and schemaVersion.

// It first checks a cache (schemaMap) for the schema, and if not found, reads it using the schemaReader.

func (engine *LookupEngine) readSchema(ctx context.Context, tenantID, schemaVersion string) (*base.SchemaDefinition, error) {

	// Create a unique cache key by combining the tenantID and schemaVersion.

	// This ensures that different combinations of tenantID and schemaVersion get their own cache entries.

	cacheKey := tenantID + "|" + schemaVersion

	// Attempt to retrieve the schema from the cache (schemaMap) using the generated cacheKey.

	if sch, ok := engine.schemaMap.Load(cacheKey); ok {

		// If the schema is present in the cache, cast it to its correct type and return.

		return sch.(*base.SchemaDefinition), nil

	}

	// If the schema is not present in the cache, use the schemaReader to read it from the source (e.g., a database or file).

	sch, err := engine.schemaReader.ReadSchema(ctx, tenantID, schemaVersion)

	if err != nil {

		// If there's an error reading the schema (e.g., schema not found or database connection issue), return the error.

		return nil, err

	}

	// Cache the newly read schema in schemaMap so that subsequent reads can be faster.

	engine.schemaMap.Store(cacheKey, sch)

	// Return the freshly read schema.

	return sch, nil

}