package org.zalando.nakadi.controller;

import org.json.JSONObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.context.request.NativeWebRequest;
import org.zalando.nakadi.domain.Storage;
import org.zalando.nakadi.exceptions.runtime.DbWriteOperationsBlockedException;
import org.zalando.nakadi.exceptions.runtime.DuplicatedStorageException;
import org.zalando.nakadi.exceptions.runtime.InternalNakadiException;
import org.zalando.nakadi.exceptions.runtime.NoSuchStorageException;
import org.zalando.nakadi.exceptions.runtime.StorageIsUsedException;
import org.zalando.nakadi.exceptions.runtime.UnknownStorageTypeException;
import org.zalando.nakadi.plugin.api.PluginException;
import org.zalando.nakadi.plugin.api.authz.AuthorizationService;
import org.zalando.nakadi.service.AdminService;
import org.zalando.nakadi.service.StorageService;

import java.util.List;

import static org.springframework.http.HttpStatus.CREATED;
import static org.springframework.http.HttpStatus.FORBIDDEN;
import static org.springframework.http.HttpStatus.NO_CONTENT;
import static org.springframework.http.HttpStatus.OK;
import static org.springframework.http.ResponseEntity.status;

@RestController
public class StoragesController {

    private final StorageService storageService;
    private final AdminService adminService;

    @Autowired
    public StoragesController(final StorageService storageService, final AdminService adminService) {
        this.storageService = storageService;
        this.adminService = adminService;
    }

    @RequestMapping(value = "/storages", method = RequestMethod.GET)
    public ResponseEntity<?> listStorages(final NativeWebRequest request)
            throws InternalNakadiException, PluginException {
        if (!adminService.isAdmin(AuthorizationService.Operation.READ)) {
            return status(FORBIDDEN).build();
        }
        final List<Storage> storages = storageService.listStorages();
        return status(OK).body(storages);
    }

    @RequestMapping(value = "/storages", method = RequestMethod.POST)
    public ResponseEntity<?> createStorage(@RequestBody final String storage,
                                           final NativeWebRequest request)
            throws PluginException, DbWriteOperationsBlockedException,
            DuplicatedStorageException, InternalNakadiException, UnknownStorageTypeException {
        if (!adminService.isAdmin(AuthorizationService.Operation.WRITE)) {
            return status(FORBIDDEN).build();
        }
        storageService.createStorage(new JSONObject(storage));
        return status(CREATED).build();
    }

    @RequestMapping(value = "/storages/{id}", method = RequestMethod.GET)
    public ResponseEntity<?> getStorage(@PathVariable("id") final String id, final NativeWebRequest request)
            throws PluginException, NoSuchStorageException, InternalNakadiException {
        if (!adminService.isAdmin(AuthorizationService.Operation.READ)) {
            return status(FORBIDDEN).build();
        }
        final Storage storage = storageService.getStorage(id);
        return status(OK).body(storage);
    }

    @RequestMapping(value = "/storages/{id}", method = RequestMethod.DELETE)
    public ResponseEntity<?> deleteStorage(@PathVariable("id") final String id, final NativeWebRequest request)
            throws PluginException, DbWriteOperationsBlockedException, NoSuchStorageException,
            StorageIsUsedException, InternalNakadiException{
        if (!adminService.isAdmin(AuthorizationService.Operation.WRITE)) {
            return status(FORBIDDEN).build();
        }
        storageService.deleteStorage(id);
        return status(NO_CONTENT).build();
    }

    @RequestMapping(value = "/storages/default/{id}", method = RequestMethod.PUT)
    public ResponseEntity<?> setDefaultStorage(@PathVariable("id") final String id, final NativeWebRequest request)
            throws PluginException, NoSuchStorageException, InternalNakadiException {
        if (!adminService.isAdmin(AuthorizationService.Operation.WRITE)) {
            return status(FORBIDDEN).build();
        }
        final Storage storage = storageService.setDefaultStorage(id);
        return status(OK).body(storage);
    }
}

@@ -733,32 +733,106 @@
 orageException;%0A
+import org.zalando.nakadi.exceptions.runtime.ForbiddenOperationException;%0A
 import org.zalan
@@ -1402,69 +1402,8 @@
 ED;%0A
-import static org.springframework.http.HttpStatus.FORBIDDEN;%0A
 impo
@@ -2129,32 +2129,61 @@
  PluginException
+, ForbiddenOperationException
  %7B%0A        if (!
@@ -2259,39 +2259,99 @@
     
-return status(FORBIDDEN).build(
+throw new ForbiddenOperationException(%22Admin privileges required to perform this operation%22
 );%0A 
@@ -2856,32 +2856,73 @@
 ageTypeException
+,%0A            ForbiddenOperationException
  %7B%0A        if (!
@@ -2999,39 +2999,99 @@
     
-return status(FORBIDDEN).build(
+throw new ForbiddenOperationException(%22Admin privileges required to perform this operation%22
 );%0A 
@@ -3467,32 +3467,61 @@
 lNakadiException
+, ForbiddenOperationException
  %7B%0A        if (!
@@ -3597,39 +3597,99 @@
     
-return status(FORBIDDEN).build(
+throw new ForbiddenOperationException(%22Admin privileges required to perform this operation%22
 );%0A 
@@ -4151,16 +4151,46 @@
 xception
+, ForbiddenOperationException 
 %7B%0A      
@@ -4274,39 +4274,99 @@
     
-return status(FORBIDDEN).build(
+throw new ForbiddenOperationException(%22Admin privileges required to perform this operation%22
 );%0A 
@@ -4739,32 +4739,61 @@
 lNakadiException
+, ForbiddenOperationException
  %7B%0A        if (!
@@ -4870,39 +4870,99 @@
     
-return status(FORBIDDEN).build(
+throw new ForbiddenOperationException(%22Admin privileges required to perform this operation%22
 );%0A 

Handle 403 consistently