// anyPrivilege implements the DescriptorAccessor interface.
func (p *planner) anyPrivilege(descriptor sqlbase.DescriptorProto) error {
	if userCanSeeDescriptor(descriptor, p.session.User) {
		return nil
	}
	return fmt.Errorf("user %s has no privileges on %s %s",
		p.session.User, descriptor.TypeName(), descriptor.GetName())
}

// createDescriptor implements the DescriptorAccessor interface.
func (p *planner) createDescriptor(
	plainKey sqlbase.DescriptorKey, descriptor sqlbase.DescriptorProto, ifNotExists bool,
) (bool, error) {
	idKey := plainKey.Key()

	if exists, err := p.descExists(idKey); err == nil && exists {
		if ifNotExists {
			// Noop.
			return false, nil
		}
		// Key exists, but we don't want it to: error out.
		switch descriptor.TypeName() {
		case "database":
			return false, sqlbase.NewDatabaseAlreadyExistsError(plainKey.Name())
		case "table", "view":
			return false, sqlbase.NewRelationAlreadyExistsError(plainKey.Name())
		default:
			return false, descriptorAlreadyExistsErr{descriptor, plainKey.Name()}
		}
	} else if err != nil {
		return false, err
	}

	id, err := GenerateUniqueDescID(p.txn)
	if err != nil {
		return false, err
	}

	return true, p.createDescriptorWithID(idKey, id, descriptor)
}

func (p *planner) createDescriptorWithID(
	idKey roachpb.Key, id sqlbase.ID, descriptor sqlbase.DescriptorProto,
) error {
	descriptor.SetID(id)
	// TODO(pmattis): The error currently returned below is likely going to be
	// difficult to interpret.
	//
	// TODO(pmattis): Need to handle if-not-exists here as well.
	//
	// TODO(pmattis): This is writing the namespace and descriptor table entries,
	// but not going through the normal INSERT logic and not performing a precise
	// mimicry. In particular, we're only writing a single key per table, while
	// perfect mimicry would involve writing a sentinel key for each row as well.
	descKey := sqlbase.MakeDescMetadataKey(descriptor.GetID())

	b := &client.Batch{}
	descID := descriptor.GetID()
	descDesc := sqlbase.WrapDescriptor(descriptor)
	if log.V(2) {
		log.Infof(p.ctx(), "CPut %s -> %d", idKey, descID)
		log.Infof(p.ctx(), "CPut %s -> %s", descKey, descDesc)
	}
	b.CPut(idKey, descID, nil)
	b.CPut(descKey, descDesc, nil)

	p.setTestingVerifyMetadata(func(systemConfig config.SystemConfig) error {
		if err := expectDescriptorID(systemConfig, idKey, descID); err != nil {
			return err
		}
		return expectDescriptor(systemConfig, descKey, descDesc)
	})

	return p.txn.Run(b)
}

// checkPrivilege implements the DescriptorAccessor interface.
func (p *planner) checkPrivilege(
	descriptor sqlbase.DescriptorProto, privilege privilege.Kind,
) error {
	if descriptor.GetPrivileges().CheckPrivilege(p.session.User, privilege) {
		return nil
	}
	return fmt.Errorf("user %s does not have %s privilege on %s %s",
		p.session.User, privilege, descriptor.TypeName(), descriptor.GetName())
}

// isVirtualDescriptor checks if the provided DescriptorProto is an instance of
// a Virtual Descriptor.
func isVirtualDescriptor(desc sqlbase.DescriptorProto) bool {
	return desc.GetID() == keys.VirtualDescriptorID
}

func userCanSeeDescriptor(descriptor sqlbase.DescriptorProto, user string) bool {
	return descriptor.GetPrivileges().AnyPrivilege(user) || isVirtualDescriptor(descriptor)
}